yuqei/ai-exchange
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 5 plan replay backtest

Browse Source
This commit is contained in:
Codex
2026-05-06 15:35:21 +08:00
parent 79e21aeee4
commit 16f319a603
12 changed files with 900 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+8 -1
View File
@@ -222,7 +222,7 @@ Record the current commit as the last-known-good SHA before deploy. If any gate
- `python3 scripts/run_signals.py` - generate bias and signals with structural `target_plan` selection; TP1 prefers IRL and TP2 prefers ERL, and signals are skipped when traceable structural targets are missing
- `python3 scripts/run_execution.py` - run paper execution gating, require structural targets plus management rules for modeled signals, create an execution record, then auto-settle the latest execution for that signal
- `python3 scripts/run_risk_governor.py` - run the independent market-state and risk governor pre-execution gate
- `python3 scripts/run_backtest.py` - run backtest and write results, including target-source traceability and management rules in result meta
- `python3 scripts/run_backtest.py` - run backtest as `plan_replay` by default, replay active-plan candidates from appearance to confirmation to entry/invalidation, and write target-source traceability plus management rules into result meta
- `python3 scripts/run_autonomous_cycle.py` - run the unattended cycle: offline checks, database probe, readiness checks, then guarded pipeline when the database is available; prints `Next actions` for the next unattended step
- `python3 scripts/run_database_readiness.py` - run read-only database readiness checks for connection, schema, seed data, and candle availability; prints `next_actions` and `action_plan` in JSON mode
- `.venv/bin/python scripts/run_trading_delivery_rehearsal.py` - run the P0 trading delivery rehearsal as one scripted flow: checklist drift check, historical repair decision, readiness/auto-backfill, guarded pipeline, deploy/health-check, and post-deploy observe
@@ -258,6 +258,7 @@ Optional utility JSON output variables:
- `TRADE_PLAN_CANDIDATES_OUTPUT_JSON`
- `TRADING_PLANS_OUTPUT_JSON`
- `TRADING_REHEARSAL_OUTPUT_JSON`
- `BACKTEST_OUTPUT_JSON`
Optional structure script JSON output variables:
@@ -402,6 +403,12 @@ Optional trading plan environment variables used by `run_trading_plans.py`:
- `TRADING_PLAN_SETUP_TIMEFRAME`
- `TRADING_PLAN_TRIGGER_TIMEFRAME`
Optional backtest replay environment variables used by `run_backtest.py`:
- `BACKTEST_SUBJECT_TYPE` - defaults to `plan_replay`; use `signal` only when you intentionally want the legacy signal-only backtest
- `BACKTEST_TRADING_PLAN_ID` - optional plan scope for replay
- `BACKTEST_CANDIDATE_MODEL_CODE` - optional candidate model filter such as `OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE`
Optional trade plan candidate environment variables used by `run_trade_plan_candidates.py`:
- `TRADE_PLAN_CANDIDATES_OUTPUT_JSON` - set to `1`, `true`, `yes`, or `on` for machine-readable candidate output
migrations/023_add_plan_replay_backtest_dimensions.sql
+24
View File
@@ -0,0 +1,24 @@
alter table if exists backtest_runs
add column if not exists subject_type text,
add column if not exists trading_plan_id bigint references trading_plans(id),
add column if not exists candidate_model_code text;
update backtest_runs
set subject_type = coalesce(subject_type, 'signal');
create index if not exists idx_backtest_runs_subject_started_desc
on backtest_runs (subject_type, started_ts desc);
create index if not exists idx_backtest_runs_plan_started_desc
on backtest_runs (trading_plan_id, started_ts desc);
alter table if exists backtest_results
add column if not exists trading_plan_id bigint references trading_plans(id),
add column if not exists candidate_id bigint references trade_plan_candidates(id),
add column if not exists candidate_model_code text;
create index if not exists idx_backtest_results_plan_id
on backtest_results (trading_plan_id, id desc);
create index if not exists idx_backtest_results_candidate_id
on backtest_results (candidate_id, id desc);
package.json
+1 -1
View File
@@ -3,7 +3,7 @@
"version": "0.1.0",
"private": true,
"scripts": {
"lint": ".\\.venv\\Scripts\\python.exe scripts/run_script_static_checks.py && .\\.venv\\Scripts\\python.exe scripts/run_migration_static_checks.py && .\\.venv\\Scripts\\python.exe -m unittest tests.unit.test_trading_plan_service tests.unit.test_market_state_and_risk_governor tests.unit.test_target_selection_service tests.unit.test_trade_plan_candidate_service tests.unit.test_execution_service tests.unit.test_api_server",
"lint": ".\\.venv\\Scripts\\python.exe scripts/run_script_static_checks.py && .\\.venv\\Scripts\\python.exe scripts/run_migration_static_checks.py && .\\.venv\\Scripts\\python.exe -m unittest tests.unit.test_trading_plan_service tests.unit.test_market_state_and_risk_governor tests.unit.test_target_selection_service tests.unit.test_trade_plan_candidate_service tests.unit.test_backtest_service tests.unit.test_backtest_script tests.unit.test_execution_service tests.unit.test_api_server tests.unit.test_api_and_main",
"build": ".\\.venv\\Scripts\\python.exe -m compileall -q src scripts tests",
"visual:snapshots": "node --check src/web/app.js"
}
scripts/run_backtest.py
+39 -2
View File
@@ -12,6 +12,7 @@ if str(ROOT) not in sys.path:
from src.services.backtest import (
BACKTEST_MODE_ENTRY_THEN_OUTCOME,
BACKTEST_SUBJECT_PLAN_REPLAY,
BacktestCostConfig,
BacktestQualityGate,
BacktestReplayConfig,
@@ -55,6 +56,26 @@ def build_backtest_mode_from_env(env=None) -> str:
return value or BACKTEST_MODE_ENTRY_THEN_OUTCOME
def build_backtest_subject_type_from_env(env=None) -> str:
env = os.environ if env is None else env
value = str(env.get("BACKTEST_SUBJECT_TYPE") or "").strip()
return value or BACKTEST_SUBJECT_PLAN_REPLAY
def build_backtest_trading_plan_id_from_env(env=None) -> Optional[int]:
env = os.environ if env is None else env
value = env.get("BACKTEST_TRADING_PLAN_ID")
if value is None or value == "":
return None
return int(value)
def build_backtest_candidate_model_code_from_env(env=None) -> Optional[str]:
env = os.environ if env is None else env
value = str(env.get("BACKTEST_CANDIDATE_MODEL_CODE") or "").strip()
return value or None
def build_replay_config_from_env(env=None) -> BacktestReplayConfig:
env = os.environ if env is None else env
defaults = BacktestReplayConfig()
@@ -85,6 +106,9 @@ def run_backtest_once(
require_quality_gate: bool = False,
output_json: bool = False,
backtest_mode: str = BACKTEST_MODE_ENTRY_THEN_OUTCOME,
subject_type: str = BACKTEST_SUBJECT_PLAN_REPLAY,
trading_plan_id: Optional[int] = None,
candidate_model_code: Optional[str] = None,
replay_config: Optional[BacktestReplayConfig] = None,
timeframe_set: Optional[dict] = None,
) -> int:
@@ -93,16 +117,26 @@ def run_backtest_once(
limit=limit,
cost_config=cost_config,
quality_gate=quality_gate,
subject_type=subject_type,
trading_plan_id=trading_plan_id,
candidate_model_code=candidate_model_code,
mode=backtest_mode,
replay_config=replay_config,
timeframe_set=timeframe_set,
)
summary = summarize_backtest_outcomes(outcomes)
quality_gate_result = evaluate_backtest_quality(summary, gate=quality_gate)
summary = summarize_backtest_outcomes(outcomes, subject_type=subject_type)
quality_gate_result = evaluate_backtest_quality(
summary,
gate=quality_gate,
sample_basis="executed" if subject_type == BACKTEST_SUBJECT_PLAN_REPLAY else "total_plus_executed",
)
payload = {
"passed": not require_quality_gate or quality_gate_result["passed"],
"backtest_run_id": run_id,
"result_count": len(outcomes),
"subject_type": subject_type,
"trading_plan_id": trading_plan_id,
"candidate_model_code": candidate_model_code,
"timeframe_set": timeframe_set,
"summary": summary,
"quality_gate": quality_gate_result,
@@ -133,6 +167,9 @@ def main(env=None) -> int:
require_quality_gate=build_require_quality_gate_from_env(env=env),
output_json=output_json,
backtest_mode=build_backtest_mode_from_env(env=env),
subject_type=build_backtest_subject_type_from_env(env=env),
trading_plan_id=build_backtest_trading_plan_id_from_env(env=env),
candidate_model_code=build_backtest_candidate_model_code_from_env(env=env),
replay_config=build_replay_config_from_env(env=env),
timeframe_set=build_backtest_timeframe_set_from_env(env=env),
)
src/main.py
+5
View File
@@ -5,6 +5,8 @@ from typing import Optional
from src.api import get_health, get_recent_backtest_summaries, get_recent_signals, get_recent_trade_plan_candidates
from src.services.backtest import (
BACKTEST_MODE_ENTRY_THEN_OUTCOME,
BACKTEST_SUBJECT_PLAN_REPLAY,
BACKTEST_SUBJECT_SIGNAL,
BacktestCostConfig,
BacktestQualityGate,
BacktestReplayConfig,
@@ -268,6 +270,9 @@ def run_closed_loop(
limit=backtest_limit,
cost_config=backtest_cost_config,
quality_gate=backtest_quality_gate,
subject_type=BACKTEST_SUBJECT_PLAN_REPLAY if latest_trade_plan_candidate is not None else BACKTEST_SUBJECT_SIGNAL,
trading_plan_id=(latest_trade_plan_candidate or {}).get("trading_plan_id") if latest_trade_plan_candidate is not None else None,
candidate_model_code=(latest_trade_plan_candidate or {}).get("model_code") if latest_trade_plan_candidate is not None else None,
mode=backtest_mode,
replay_config=backtest_replay_config,
timeframe_set=timeframe_set,
src/models/backtest_result.py
+3
View File
@@ -11,6 +11,9 @@ class BacktestResult(Base):
id: Mapped[int] = mapped_column(primary_key=True)
backtest_run_id: Mapped[int] = mapped_column(ForeignKey("backtest_runs.id"), nullable=False)
trading_plan_id: Mapped[Optional[int]] = mapped_column(ForeignKey("trading_plans.id"))
candidate_id: Mapped[Optional[int]] = mapped_column(ForeignKey("trade_plan_candidates.id"))
candidate_model_code: Mapped[Optional[str]] = mapped_column(Text)
signal_id: Mapped[Optional[int]] = mapped_column(ForeignKey("trade_signals.id"))
outcome: Mapped[str] = mapped_column(Text, nullable=False)
mfe: Mapped[Optional[float]] = mapped_column(Numeric(20, 10))
src/models/backtest_run.py
+3
View File
@@ -10,8 +10,11 @@ class BacktestRun(Base):
__tablename__ = "backtest_runs"
id: Mapped[int] = mapped_column(primary_key=True)
subject_type: Mapped[Optional[str]] = mapped_column(Text)
model_code: Mapped[str] = mapped_column(Text, nullable=False)
instrument_id: Mapped[Optional[int]] = mapped_column(ForeignKey("instruments.id"))
trading_plan_id: Mapped[Optional[int]] = mapped_column(ForeignKey("trading_plans.id"))
candidate_model_code: Mapped[Optional[str]] = mapped_column(Text)
timeframe_set: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
started_ts: Mapped[object] = mapped_column(nullable=False)
ended_ts: Mapped[Optional[object]] = mapped_column()
src/repositories/backtest_repository.py
+129 -2
View File
@@ -1,5 +1,6 @@
from datetime import datetime, timezone
import json
from typing import Optional
from sqlalchemy import text
@@ -7,6 +8,37 @@ from src.db.session import SessionLocal
class BacktestRepository:
def fetch_active_trading_plan(self, instrument_id: int, trading_plan_id: int = None) -> Optional[dict]:
if trading_plan_id is not None:
sql = text(
"""
select id, instrument_id, plan_date, status, narrative, primary_draw, secondary_draw,
bias, allowed_sessions, invalidations, no_trade_conditions, inputs, quality,
timeframe_set, created_ts, activated_ts
from trading_plans
where id = :trading_plan_id
limit 1
"""
)
params = {"trading_plan_id": trading_plan_id}
else:
sql = text(
"""
select id, instrument_id, plan_date, status, narrative, primary_draw, secondary_draw,
bias, allowed_sessions, invalidations, no_trade_conditions, inputs, quality,
timeframe_set, created_ts, activated_ts
from trading_plans
where instrument_id = :instrument_id
and status = 'active'
order by activated_ts desc nulls last, plan_date desc, id desc
limit 1
"""
)
params = {"instrument_id": instrument_id}
with SessionLocal() as session:
row = session.execute(sql, params).mappings().first()
return dict(row) if row else None
def fetch_candidate_signals(self, instrument_id: int, limit: int = 100, timeframe_set: dict = None) -> list[dict]:
timeframe_set = dict(timeframe_set or {})
filters = ["ts.instrument_id = :instrument_id"]
@@ -48,6 +80,70 @@ class BacktestRepository:
rows = session.execute(sql, params).mappings().all()
return [dict(row) for row in rows]
def fetch_plan_replay_candidate_rows(
self,
instrument_id: int,
limit: int = 100,
timeframe_set: dict = None,
trading_plan_id: int = None,
candidate_model_code: str = None,
) -> list[dict]:
timeframe_set = dict(timeframe_set or {})
filters = ["tpc.instrument_id = :instrument_id"]
params = {"instrument_id": instrument_id, "limit": limit}
if trading_plan_id is not None:
filters.append("tpc.trading_plan_id = :trading_plan_id")
params["trading_plan_id"] = trading_plan_id
if candidate_model_code:
filters.append("tpc.model_code = :candidate_model_code")
params["candidate_model_code"] = candidate_model_code
if timeframe_set.get("setup_tf"):
filters.append("tpc.setup_timeframe = :setup_tf")
params["setup_tf"] = timeframe_set["setup_tf"]
if timeframe_set.get("trigger_tf"):
filters.append("tpc.trigger_timeframe = :trigger_tf")
params["trigger_tf"] = timeframe_set["trigger_tf"]
where_sql = " and ".join(filters)
sql = text(
f"""
select
tpc.id as candidate_id,
tpc.instrument_id,
tpc.trading_plan_id,
tpc.model_code as candidate_model_code,
tpc.status as candidate_status,
tpc.side,
tpc.setup_timeframe,
tpc.trigger_timeframe,
tpc.created_ts,
tpc.expires_ts,
tpc.confirmation_state,
tpc.missing_confirmations,
tpc.evidence,
tpc.entry,
tpc.invalidations,
tpc.narrative,
tpc.meta as candidate_meta,
tp.status as trading_plan_status,
tp.plan_date,
tp.allowed_sessions,
tp.invalidations as trading_plan_invalidations,
tp.primary_draw,
tp.secondary_draw,
tp.bias,
tp.activated_ts,
tp.created_ts as trading_plan_created_ts
from trade_plan_candidates tpc
join trading_plans tp on tp.id = tpc.trading_plan_id
where {where_sql}
order by tpc.created_ts asc, tpc.id asc
limit :limit
"""
)
with SessionLocal() as session:
rows = session.execute(sql, params).mappings().all()
return [dict(row) for row in rows]
def fetch_future_candles(self, instrument_id: int, timeframe: str, start_ts, end_ts, limit: int = 500) -> list[dict]:
sql = text(
"""
@@ -92,19 +188,34 @@ class BacktestRepository:
).mappings().all()
return [dict(row) for row in rows]
def insert_backtest_run(self, model_code: str, instrument_id: int, timeframe_set: dict, config: dict) -> int:
def insert_backtest_run(
self,
model_code: str,
instrument_id: int,
timeframe_set: dict,
config: dict,
subject_type: str = "signal",
trading_plan_id: int = None,
candidate_model_code: str = None,
) -> int:
sql = text(
"""
insert into backtest_runs (
subject_type,
model_code,
instrument_id,
trading_plan_id,
candidate_model_code,
timeframe_set,
started_ts,
config,
summary
) values (
:subject_type,
:model_code,
:instrument_id,
:trading_plan_id,
:candidate_model_code,
cast(:timeframe_set as jsonb),
:started_ts,
cast(:config as jsonb),
@@ -117,8 +228,11 @@ class BacktestRepository:
result = session.execute(
sql,
{
"subject_type": subject_type,
"model_code": model_code,
"instrument_id": instrument_id,
"trading_plan_id": trading_plan_id,
"candidate_model_code": candidate_model_code,
"timeframe_set": json.dumps(timeframe_set),
"started_ts": datetime.now(timezone.utc),
"config": json.dumps(config),
@@ -128,8 +242,12 @@ class BacktestRepository:
def insert_backtest_result(
self,
*,
backtest_run_id: int,
signal_id: int,
signal_id: int = None,
trading_plan_id: int = None,
candidate_id: int = None,
candidate_model_code: str = None,
outcome: str,
mfe: float,
mae: float,
@@ -143,6 +261,9 @@ class BacktestRepository:
"""
insert into backtest_results (
backtest_run_id,
trading_plan_id,
candidate_id,
candidate_model_code,
signal_id,
outcome,
mfe,
@@ -154,6 +275,9 @@ class BacktestRepository:
meta
) values (
:backtest_run_id,
:trading_plan_id,
:candidate_id,
:candidate_model_code,
:signal_id,
:outcome,
:mfe,
@@ -171,6 +295,9 @@ class BacktestRepository:
sql,
{
"backtest_run_id": backtest_run_id,
"trading_plan_id": trading_plan_id,
"candidate_id": candidate_id,
"candidate_model_code": candidate_model_code,
"signal_id": signal_id,
"outcome": outcome,
"mfe": mfe,
src/services/backtest/__init__.py
+6
View File
@@ -2,6 +2,8 @@
from src.services.backtest.backtest_service import (
BACKTEST_MODE_ENTRY_THEN_OUTCOME,
BACKTEST_SUBJECT_PLAN_REPLAY,
BACKTEST_SUBJECT_SIGNAL,
BACKTEST_LIFECYCLE_AWAITING_CONFIRMATION,
BACKTEST_LIFECYCLE_INVALIDATED_BEFORE_ENTRY,
BACKTEST_LIFECYCLE_MISSED,
@@ -18,6 +20,7 @@ from src.services.backtest.backtest_service import (
BacktestQualityGate,
BacktestSignal,
build_cumulative_r_curve,
build_plan_replay_units,
build_timeframe_bucket,
compute_max_consecutive_losses,
compute_max_drawdown_r,
@@ -45,6 +48,8 @@ from src.services.backtest.replay import (
__all__ = [
"BacktestOutcome",
"BACKTEST_MODE_ENTRY_THEN_OUTCOME",
"BACKTEST_SUBJECT_PLAN_REPLAY",
"BACKTEST_SUBJECT_SIGNAL",
"BACKTEST_LIFECYCLE_AWAITING_CONFIRMATION",
"BACKTEST_LIFECYCLE_INVALIDATED_BEFORE_ENTRY",
"BACKTEST_LIFECYCLE_MISSED",
@@ -60,6 +65,7 @@ __all__ = [
"BacktestQualityGate",
"BacktestSignal",
"build_cumulative_r_curve",
"build_plan_replay_units",
"build_timeframe_bucket",
"compute_max_consecutive_losses",
"compute_max_drawdown_r",
src/services/backtest/backtest_service.py
+465 -10
View File
@@ -7,8 +7,14 @@ from src.services.backtest.replay import ReplaySwing, iter_historical_replay
SAME_CANDLE_PRIORITY = "stop_first"
BACKTEST_SUBJECT_SIGNAL = "signal"
BACKTEST_SUBJECT_PLAN_REPLAY = "plan_replay"
BACKTEST_MODE_ENTRY_THEN_OUTCOME = "entry_then_outcome"
BACKTEST_MODE_REPLAY_NO_LOOKAHEAD = "replay_no_lookahead"
SUPPORTED_BACKTEST_SUBJECT_TYPES = {
BACKTEST_SUBJECT_SIGNAL,
BACKTEST_SUBJECT_PLAN_REPLAY,
}
SUPPORTED_BACKTEST_MODES = {
BACKTEST_MODE_ENTRY_THEN_OUTCOME,
BACKTEST_MODE_REPLAY_NO_LOOKAHEAD,
@@ -35,6 +41,9 @@ class BacktestOutcome:
invalidated_before_entry: bool
error_tags: list[str]
meta: dict
trading_plan_id: Optional[int] = None
candidate_id: Optional[int] = None
candidate_model_code: Optional[str] = None
@dataclass(frozen=True)
@@ -50,6 +59,11 @@ class BacktestSignal:
rr_tp2: float = 0.0
target_plan: Optional[dict] = None
management_rules: Optional[list[dict]] = None
trading_plan_id: Optional[int] = None
candidate_id: Optional[int] = None
candidate_model_code: Optional[str] = None
candidate_appeared_ts: Optional[object] = None
candidate_confirmed_ts: Optional[object] = None
@dataclass(frozen=True)
@@ -85,6 +99,9 @@ class BacktestService:
limit: int = 100,
cost_config: Optional[BacktestCostConfig] = None,
quality_gate: Optional[BacktestQualityGate] = None,
subject_type: str = BACKTEST_SUBJECT_SIGNAL,
trading_plan_id: Optional[int] = None,
candidate_model_code: Optional[str] = None,
mode: str = BACKTEST_MODE_ENTRY_THEN_OUTCOME,
replay_config: Optional[BacktestReplayConfig] = None,
timeframe_set: Optional[dict] = None,
@@ -93,20 +110,26 @@ class BacktestService:
quality_gate = quality_gate or BacktestQualityGate()
replay_config = replay_config or BacktestReplayConfig()
timeframe_set = normalize_timeframe_set(timeframe_set)
if subject_type not in SUPPORTED_BACKTEST_SUBJECT_TYPES:
raise ValueError(f"invalid_backtest_subject_type:{subject_type}")
if mode not in SUPPORTED_BACKTEST_MODES:
raise ValueError(f"invalid_backtest_mode:{mode}")
signals = self.repository.fetch_candidate_signals(
instrument_id=instrument_id,
limit=limit,
timeframe_set=timeframe_set,
)
if subject_type == BACKTEST_SUBJECT_PLAN_REPLAY and trading_plan_id is None:
active_trading_plan = self.repository.fetch_active_trading_plan(instrument_id=instrument_id)
trading_plan_id = active_trading_plan.get('id') if active_trading_plan else None
run_id = self.repository.insert_backtest_run(
model_code='SWEEP_MSS_FVG',
instrument_id=instrument_id,
timeframe_set=timeframe_set,
subject_type=subject_type,
trading_plan_id=trading_plan_id,
candidate_model_code=candidate_model_code,
config={
'limit': limit,
'subject_type': subject_type,
'mode': mode,
'trading_plan_id': trading_plan_id,
'candidate_model_code': candidate_model_code,
'same_candle_priority': 'stop_first',
'session_filter': 'mark_outside',
'cost_model': cost_config.__dict__,
@@ -114,6 +137,28 @@ class BacktestService:
'replay': replay_config.__dict__ if mode == BACKTEST_MODE_REPLAY_NO_LOOKAHEAD else None,
},
)
if subject_type == BACKTEST_SUBJECT_PLAN_REPLAY:
outcomes = self._run_plan_replay(
run_id=run_id,
instrument_id=instrument_id,
limit=limit,
trading_plan_id=trading_plan_id,
candidate_model_code=candidate_model_code,
cost_config=cost_config,
mode=mode,
replay_config=replay_config,
timeframe_set=timeframe_set,
)
summary = summarize_backtest_outcomes(outcomes, subject_type=subject_type)
summary['quality_gate'] = evaluate_backtest_quality(summary, gate=quality_gate, sample_basis='executed')
self.repository.update_backtest_run_summary(run_id, summary)
return run_id, outcomes
signals = self.repository.fetch_candidate_signals(
instrument_id=instrument_id,
limit=limit,
timeframe_set=timeframe_set,
)
outcomes: list[BacktestOutcome] = []
for signal in signals:
sessions = self.repository.fetch_sessions_covering_window(
@@ -176,6 +221,9 @@ class BacktestService:
self.repository.insert_backtest_result(
backtest_run_id=run_id,
signal_id=record.signal_id,
trading_plan_id=record.trading_plan_id,
candidate_id=record.candidate_id,
candidate_model_code=record.candidate_model_code,
outcome=record.outcome,
mfe=record.mfe,
mae=record.mae,
@@ -201,6 +249,9 @@ class BacktestService:
rr_tp2=float(signal.get('rr_tp2') or 0),
target_plan=extract_target_plan_from_signal_row(signal),
management_rules=extract_management_rules_from_signal_row(signal),
trading_plan_id=signal.get('trading_plan_id'),
candidate_id=signal.get('candidate_id'),
candidate_model_code=signal.get('candidate_model_code'),
),
future,
cost_config=cost_config,
@@ -238,6 +289,9 @@ class BacktestService:
self.repository.insert_backtest_result(
backtest_run_id=run_id,
signal_id=record.signal_id,
trading_plan_id=record.trading_plan_id,
candidate_id=record.candidate_id,
candidate_model_code=record.candidate_model_code,
outcome=record.outcome,
mfe=record.mfe,
mae=record.mae,
@@ -247,11 +301,195 @@ class BacktestService:
error_tags=record.error_tags,
meta=record.meta,
)
summary = summarize_backtest_outcomes(outcomes)
summary = summarize_backtest_outcomes(outcomes, subject_type=subject_type)
summary['quality_gate'] = evaluate_backtest_quality(summary, gate=quality_gate)
self.repository.update_backtest_run_summary(run_id, summary)
return run_id, outcomes
def _run_plan_replay(
self,
*,
run_id: int,
instrument_id: int,
limit: int,
trading_plan_id: Optional[int],
candidate_model_code: Optional[str],
cost_config: BacktestCostConfig,
mode: str,
replay_config: BacktestReplayConfig,
timeframe_set: dict,
) -> list[BacktestOutcome]:
candidate_rows = self.repository.fetch_plan_replay_candidate_rows(
instrument_id=instrument_id,
limit=limit,
timeframe_set=timeframe_set,
trading_plan_id=trading_plan_id,
candidate_model_code=candidate_model_code,
)
replay_units = build_plan_replay_units(candidate_rows)
outcomes: list[BacktestOutcome] = []
for unit in replay_units:
record = self._replay_candidate_unit(
unit=unit,
instrument_id=instrument_id,
cost_config=cost_config,
mode=mode,
replay_config=replay_config,
timeframe_set=timeframe_set,
)
outcomes.append(record)
self.repository.insert_backtest_result(
backtest_run_id=run_id,
signal_id=record.signal_id,
trading_plan_id=record.trading_plan_id,
candidate_id=record.candidate_id,
candidate_model_code=record.candidate_model_code,
outcome=record.outcome,
mfe=record.mfe,
mae=record.mae,
tp1_hit=record.tp1_hit,
tp2_hit=record.tp2_hit,
invalidated_before_entry=record.invalidated_before_entry,
error_tags=record.error_tags,
meta=record.meta,
)
return outcomes
def _replay_candidate_unit(
self,
*,
unit: dict,
instrument_id: int,
cost_config: BacktestCostConfig,
mode: str,
replay_config: BacktestReplayConfig,
timeframe_set: dict,
) -> BacktestOutcome:
sessions = self.repository.fetch_sessions_covering_window(
instrument_id=instrument_id,
start_ts=unit['appeared_ts'],
end_ts=unit['expires_ts'],
)
if unit['confirmed_ts'] is None:
failure_reasons = list(unit['reason_codes'] or unit['missing_confirmations'] or ['candidate_not_confirmed'])
return BacktestOutcome(
signal_id=0,
trading_plan_id=unit['trading_plan_id'],
candidate_id=unit['candidate_id'],
candidate_model_code=unit['candidate_model_code'],
outcome='missed',
mfe=0.0,
mae=0.0,
tp1_hit=False,
tp2_hit=False,
invalidated_before_entry=True,
error_tags=['candidate_not_confirmed'],
meta=build_plan_replay_meta(
unit=unit,
sessions=sessions,
entry_triggered=False,
failure_reasons=failure_reasons,
timeframe_set=timeframe_set,
model_compliant=False,
mode=mode,
),
)
future = self.repository.fetch_future_candles(
instrument_id=instrument_id,
timeframe=str(unit.get('trigger_timeframe') or timeframe_set['trigger_tf']),
start_ts=unit['confirmed_ts'],
end_ts=unit['expires_ts'],
limit=500,
)
if not future:
failure_reasons = ['no_future_candles']
return BacktestOutcome(
signal_id=0,
trading_plan_id=unit['trading_plan_id'],
candidate_id=unit['candidate_id'],
candidate_model_code=unit['candidate_model_code'],
outcome='missed',
mfe=0.0,
mae=0.0,
tp1_hit=False,
tp2_hit=False,
invalidated_before_entry=True,
error_tags=failure_reasons,
meta=build_plan_replay_meta(
unit=unit,
sessions=sessions,
entry_triggered=False,
failure_reasons=failure_reasons,
timeframe_set=timeframe_set,
model_compliant=True,
mode=mode,
),
)
signal = BacktestSignal(
signal_id=0,
side=unit['side'],
entry_low=float(unit['entry_low'] or 0.0),
entry_high=float(unit['entry_high'] or 0.0),
stop_loss=float(unit['stop_loss'] or 0.0),
tp1=float(unit['tp1'] or 0.0),
tp2=float(unit['tp2'] or 0.0),
rr_tp1=float(unit['rr_tp1'] or 0.0),
rr_tp2=float(unit['rr_tp2'] or 0.0),
target_plan=dict(unit['target_plan'] or {}),
management_rules=list(unit['management_rules'] or []),
trading_plan_id=unit['trading_plan_id'],
candidate_id=unit['candidate_id'],
candidate_model_code=unit['candidate_model_code'],
candidate_appeared_ts=unit['appeared_ts'],
candidate_confirmed_ts=unit['confirmed_ts'],
)
record = evaluate_signal_outcome(
signal=signal,
future_candles=future,
cost_config=cost_config,
mode=mode,
replay_config=replay_config,
)
invalidated_before_entry = bool(
(not record.meta.get('entry_triggered') and unit.get('invalidated_ts'))
or record.invalidated_before_entry
)
failure_reasons = list(unit['reason_codes'] or [])
if record.error_tags:
failure_reasons.extend(record.error_tags)
meta = {
**record.meta,
**build_plan_replay_meta(
unit=unit,
sessions=sessions,
entry_triggered=bool(record.meta.get('entry_triggered')),
failure_reasons=_dedupe_text_list(failure_reasons),
timeframe_set=timeframe_set,
model_compliant=True,
mode=mode,
),
}
candidate_entry_ts = record.meta.get('entry_ts') or unit.get('entry_ts')
if candidate_entry_ts is not None:
meta['candidate_entry_ts'] = candidate_entry_ts
meta['draw_target_hit'] = bool(record.tp2_hit)
return BacktestOutcome(
signal_id=0,
trading_plan_id=unit['trading_plan_id'],
candidate_id=unit['candidate_id'],
candidate_model_code=unit['candidate_model_code'],
outcome=record.outcome,
mfe=record.mfe,
mae=record.mae,
tp1_hit=record.tp1_hit,
tp2_hit=record.tp2_hit,
invalidated_before_entry=invalidated_before_entry,
error_tags=_dedupe_text_list(record.error_tags),
meta=meta,
)
def evaluate_signal_outcome(
signal: BacktestSignal,
@@ -501,6 +739,8 @@ def _evaluate_signal_outcome_steps(
meta = {
'entry_triggered': entry_triggered,
'entry_ts': _normalize_meta_ts(entry_step.ts) if entry_step else None,
'exit_ts': _normalize_meta_ts(exit_step.ts) if exit_step else None,
'same_candle_priority': SAME_CANDLE_PRIORITY,
'same_candle_ambiguous': same_candle_ambiguous,
'signal_side': signal.side,
@@ -568,6 +808,181 @@ def _normalize_meta_ts(value):
return value
def build_plan_replay_units(candidate_rows: list[dict]) -> list[dict]:
grouped: dict[tuple, list[dict]] = {}
for row in candidate_rows:
key = (
row.get('trading_plan_id'),
row.get('candidate_model_code'),
row.get('side'),
row.get('setup_timeframe'),
row.get('trigger_timeframe'),
)
grouped.setdefault(key, []).append(row)
units: list[dict] = []
for rows in grouped.values():
ordered = sorted(rows, key=lambda item: (item.get('created_ts'), item.get('candidate_id') or 0))
first = ordered[0]
latest = ordered[-1]
confirmed_row = next((row for row in ordered if candidate_row_is_executable(row)), None)
blocked_row = next(
(
row
for row in ordered
if confirmed_row is not None
and row.get('created_ts') > confirmed_row.get('created_ts')
and candidate_row_is_blocked(row)
),
None,
)
entry_payload = dict((confirmed_row or latest).get('entry') or {})
target_plan = dict(
entry_payload.get('target_plan')
or dict((latest.get('candidate_meta') or {}).get('target_plan') or {})
)
management_rules = list(
entry_payload.get('management_rules')
or list((latest.get('candidate_meta') or {}).get('management_rules') or [])
)
units.append(
{
'trading_plan_id': first.get('trading_plan_id'),
'candidate_id': (confirmed_row or first).get('candidate_id'),
'candidate_model_code': first.get('candidate_model_code'),
'side': first.get('side'),
'setup_timeframe': first.get('setup_timeframe'),
'trigger_timeframe': first.get('trigger_timeframe'),
'appeared_ts': first.get('created_ts'),
'confirmed_ts': confirmed_row.get('created_ts') if confirmed_row else None,
'invalidated_ts': blocked_row.get('created_ts') if blocked_row else None,
'expires_ts': (confirmed_row or latest).get('expires_ts') or latest.get('created_ts'),
'entry_low': entry_payload.get('entry_low'),
'entry_high': entry_payload.get('entry_high'),
'stop_loss': entry_payload.get('stop_loss'),
'tp1': entry_payload.get('tp1'),
'tp2': entry_payload.get('tp2'),
'rr_tp1': derive_rr_from_entry(
side=first.get('side'),
entry_low=entry_payload.get('entry_low'),
entry_high=entry_payload.get('entry_high'),
stop_loss=entry_payload.get('stop_loss'),
target_price=entry_payload.get('tp1'),
),
'rr_tp2': derive_rr_from_entry(
side=first.get('side'),
entry_low=entry_payload.get('entry_low'),
entry_high=entry_payload.get('entry_high'),
stop_loss=entry_payload.get('stop_loss'),
target_price=entry_payload.get('tp2'),
),
'target_plan': target_plan,
'management_rules': management_rules,
'missing_confirmations': list(latest.get('missing_confirmations') or []),
'reason_codes': list(
((latest.get('confirmation_state') or {}).get('reason_codes'))
or list((latest.get('candidate_meta') or {}).get('reason_codes') or [])
),
'candidate_status': latest.get('candidate_status'),
'allowed_sessions': list(first.get('allowed_sessions') or []),
'primary_draw': dict(first.get('primary_draw') or {}),
'plan_activated_ts': first.get('activated_ts'),
'plan_created_ts': first.get('trading_plan_created_ts'),
}
)
return units
def candidate_row_is_executable(row: dict) -> bool:
confirmation_state = dict(row.get('confirmation_state') or {})
return bool(confirmation_state.get('executable')) or row.get('candidate_status') == 'executable'
def candidate_row_is_blocked(row: dict) -> bool:
return row.get('candidate_status') == 'blocked'
def derive_rr_from_entry(side, entry_low, entry_high, stop_loss, target_price) -> float:
values = [_float_from_meta(entry_low), _float_from_meta(entry_high), _float_from_meta(stop_loss), _float_from_meta(target_price)]
if any(value is None for value in values):
return 0.0
entry_price = (_float_from_meta(entry_low) + _float_from_meta(entry_high)) / 2
risk_distance = abs(entry_price - _float_from_meta(stop_loss))
if risk_distance <= 0:
return 0.0
return abs(_float_from_meta(target_price) - entry_price) / risk_distance
def build_plan_replay_meta(
*,
unit: dict,
sessions: list[dict],
entry_triggered: bool,
failure_reasons: list[str],
timeframe_set: dict,
model_compliant: bool,
mode: str,
) -> dict:
session_codes = [row.get('session_code') for row in sessions if row.get('session_code')] or ['OFF_HOURS']
return {
'subject_type': BACKTEST_SUBJECT_PLAN_REPLAY,
'trading_plan_id': unit.get('trading_plan_id'),
'plan_activated_ts': _normalize_meta_ts(unit.get('plan_activated_ts')),
'plan_created_ts': _normalize_meta_ts(unit.get('plan_created_ts')),
'candidate_id': unit.get('candidate_id'),
'candidate_model_code': unit.get('candidate_model_code'),
'candidate_appeared_ts': _normalize_meta_ts(unit.get('appeared_ts')),
'candidate_confirmed_ts': _normalize_meta_ts(unit.get('confirmed_ts')),
'candidate_invalidated_ts': _normalize_meta_ts(unit.get('invalidated_ts')),
'candidate_entry_ts': _normalize_meta_ts(unit.get('entry_ts')),
'candidate_status': unit.get('candidate_status'),
'candidate_confirmation_state': 'confirmed' if unit.get('confirmed_ts') else 'awaiting_confirmation',
'candidate_confirmed': bool(unit.get('confirmed_ts')),
'model_compliant': model_compliant,
'failure_reasons': failure_reasons,
'target_plan': dict(unit.get('target_plan') or {}),
'management_rules': list(unit.get('management_rules') or []),
'draw_target_hit': False,
'sessions_found': len(sessions),
'session_codes': session_codes,
'session_filter': 'outside_window' if len(sessions) == 0 else 'covered',
'signal_side': unit.get('side'),
'signal_symbol': f"plan:{unit.get('trading_plan_id')}",
'signal_timeframe_set': {
'bias_tf': timeframe_set.get('bias_tf'),
'setup_tf': unit.get('setup_timeframe') or timeframe_set.get('setup_tf'),
'trigger_tf': unit.get('trigger_timeframe') or timeframe_set.get('trigger_tf'),
},
'signal_timeframe_bucket': build_timeframe_bucket(
bias_tf=str(timeframe_set.get('bias_tf') or '1m'),
setup_tf=str(unit.get('setup_timeframe') or timeframe_set.get('setup_tf') or '1m'),
trigger_tf=str(unit.get('trigger_timeframe') or timeframe_set.get('trigger_tf') or '1m'),
),
'signal_created_ts': _normalize_meta_ts(unit.get('confirmed_ts') or unit.get('appeared_ts')),
'backtest_mode': mode,
'entry_triggered': entry_triggered,
'weekday': _extract_weekday(unit.get('confirmed_ts') or unit.get('appeared_ts')),
}
def _dedupe_text_list(values: list[str]) -> list[str]:
seen = set()
result = []
for value in values:
text = str(value or '').strip()
if not text or text in seen:
continue
seen.add(text)
result.append(text)
return result
def _float_from_meta(value) -> Optional[float]:
if value is None:
return None
return float(value)
def estimate_trade_cost_r(
signal: BacktestSignal,
outcome: str,
@@ -680,7 +1095,7 @@ def classify_backtest_lifecycle_state_from_result(backtest_result: Optional[dict
)
def summarize_backtest_outcomes(outcomes: list[BacktestOutcome]) -> dict:
def summarize_backtest_outcomes(outcomes: list[BacktestOutcome], subject_type: str = BACKTEST_SUBJECT_SIGNAL) -> dict:
total = len(outcomes)
grouped_by_outcome: dict[str, int] = {}
grouped_by_lifecycle_state: dict[str, int] = {}
@@ -695,6 +1110,12 @@ def summarize_backtest_outcomes(outcomes: list[BacktestOutcome]) -> dict:
mae_r_values: list[float] = []
executed_r_multiples: list[float] = []
executed_signals = 0
confirmed_candidates = 0
compliant_candidates = 0
draw_target_hits = 0
failure_reason_counts: dict[str, int] = {}
plan_total_r: dict[int, float] = {}
plan_seen: set[int] = set()
for outcome in outcomes:
grouped_by_outcome[outcome.outcome] = grouped_by_outcome.get(outcome.outcome, 0) + 1
@@ -709,6 +1130,18 @@ def summarize_backtest_outcomes(outcomes: list[BacktestOutcome]) -> dict:
if is_executed_outcome(outcome):
executed_signals += 1
executed_r_multiples.append(float(outcome.meta.get('r_multiple') or 0.0))
if outcome.meta.get('candidate_confirmed'):
confirmed_candidates += 1
if outcome.meta.get('model_compliant'):
compliant_candidates += 1
if outcome.meta.get('draw_target_hit'):
draw_target_hits += 1
for reason in list(outcome.meta.get('failure_reasons') or []) + list(outcome.error_tags or []):
failure_reason_counts[reason] = failure_reason_counts.get(reason, 0) + 1
trading_plan_id = getattr(outcome, 'trading_plan_id', None)
if trading_plan_id is not None:
plan_seen.add(int(trading_plan_id))
plan_total_r[int(trading_plan_id)] = plan_total_r.get(int(trading_plan_id), 0.0) + float(outcome.meta.get('r_multiple') or 0.0)
side = str(outcome.meta.get('signal_side') or 'unknown')
session_codes = outcome.meta.get('session_codes') or ['OFF_HOURS']
@@ -727,8 +1160,13 @@ def summarize_backtest_outcomes(outcomes: list[BacktestOutcome]) -> dict:
for session_code in session_codes:
_accumulate_bucket(grouped_by_session, str(session_code), outcome)
return {
plan_count = len(plan_seen)
plan_wins = sum(1 for total_r in plan_total_r.values() if total_r > 0.0)
candidate_count = total
result = {
'subject_type': subject_type,
'total_signals': total,
'candidate_count': candidate_count,
'outcomes': grouped_by_outcome,
'lifecycle_states': grouped_by_lifecycle_state,
'win_rate': _ratio(grouped_by_outcome.get('win', 0), total),
@@ -739,6 +1177,16 @@ def summarize_backtest_outcomes(outcomes: list[BacktestOutcome]) -> dict:
'executed_rate': _ratio(executed_signals, total),
'executed_expectancy_r': _average(executed_r_multiples),
'executed_profit_factor_r': _profit_factor(executed_r_multiples),
'candidate_confirmation_rate': _ratio(confirmed_candidates, candidate_count),
'confirmed_candidates': confirmed_candidates,
'model_compliant_candidates': compliant_candidates,
'model_compliance_rate': _ratio(compliant_candidates, candidate_count),
'draw_target_hit_rate': _ratio(draw_target_hits, executed_signals),
'draw_target_hits': draw_target_hits,
'plan_count': plan_count,
'plan_wins': plan_wins,
'plan_win_rate': _ratio(plan_wins, plan_count),
'failure_reasons': failure_reason_counts,
'avg_mfe': _average([outcome.mfe for outcome in outcomes]),
'avg_mae': _average([outcome.mae for outcome in outcomes]),
'avg_mfe_r': _average(mfe_r_values),
@@ -757,6 +1205,8 @@ def summarize_backtest_outcomes(outcomes: list[BacktestOutcome]) -> dict:
'by_symbol': grouped_by_symbol,
'by_timeframe': grouped_by_timeframe,
}
result['session_performance'] = result['by_session']
return result
def normalize_timeframe_set(timeframe_set: Optional[dict]) -> dict:
@@ -786,7 +1236,11 @@ def build_timeframe_bucket(*, bias_tf: str, setup_tf: str, trigger_tf: str) -> s
return f"{bias_tf}/{setup_tf}/{trigger_tf}"
def evaluate_backtest_quality(summary: dict, gate: Optional[BacktestQualityGate] = None) -> dict[str, object]:
def evaluate_backtest_quality(
summary: dict,
gate: Optional[BacktestQualityGate] = None,
sample_basis: str = "total_plus_executed",
) -> dict[str, object]:
gate = gate or BacktestQualityGate()
reasons: list[str] = []
@@ -797,7 +1251,7 @@ def evaluate_backtest_quality(summary: dict, gate: Optional[BacktestQualityGate]
max_consecutive_losses = int(summary.get('max_consecutive_losses') or 0)
executed_signals = int(summary.get('executed_signals') or 0)
if total_signals < gate.min_total_signals:
if sample_basis != "executed" and total_signals < gate.min_total_signals:
reasons.append('sample_size_too_small')
if executed_signals < gate.min_executed_signals:
reasons.append('executed_sample_too_small')
@@ -814,6 +1268,7 @@ def evaluate_backtest_quality(summary: dict, gate: Optional[BacktestQualityGate]
'passed': len(reasons) == 0,
'reasons': reasons,
'gate': gate.__dict__,
'sample_basis': sample_basis,
'metrics': {
'total_signals': total_signals,
'executed_signals': executed_signals,
tests/unit/test_backtest_script.py
+28 -2
View File
@@ -11,12 +11,15 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:
from scripts import run_backtest
from scripts.run_backtest import (
build_backtest_candidate_model_code_from_env,
build_backtest_timeframe_set_from_env,
build_backtest_mode_from_env,
build_backtest_subject_type_from_env,
build_cost_config_from_env,
build_quality_gate_from_env,
build_replay_config_from_env,
build_require_quality_gate_from_env,
build_backtest_trading_plan_id_from_env,
run_backtest_once,
)
from src.services.backtest import BacktestOutcome
@@ -91,6 +94,12 @@ class RunBacktestScriptTests(unittest.TestCase):
def test_build_backtest_mode_and_replay_config_from_env(self) -> None:
self.assertEqual(build_backtest_mode_from_env({}), "entry_then_outcome")
self.assertEqual(build_backtest_mode_from_env({"BACKTEST_MODE": "replay_no_lookahead"}), "replay_no_lookahead")
self.assertEqual(build_backtest_subject_type_from_env({}), "plan_replay")
self.assertEqual(build_backtest_subject_type_from_env({"BACKTEST_SUBJECT_TYPE": "signal"}), "signal")
self.assertIsNone(build_backtest_trading_plan_id_from_env({}))
self.assertEqual(build_backtest_trading_plan_id_from_env({"BACKTEST_TRADING_PLAN_ID": "77"}), 77)
self.assertIsNone(build_backtest_candidate_model_code_from_env({}))
self.assertEqual(build_backtest_candidate_model_code_from_env({"BACKTEST_CANDIDATE_MODEL_CODE": "OSOK"}), "OSOK")
replay_config = build_replay_config_from_env(
{
@@ -138,7 +147,7 @@ class RunBacktestScriptTests(unittest.TestCase):
exit_code = run_backtest_once(service=FakeBacktestService([]), require_quality_gate=True)
self.assertEqual(exit_code, 2)
self.assertIn("sample_size_too_small", output.getvalue())
self.assertIn("executed_sample_too_small", output.getvalue())
def test_run_backtest_once_passes_when_required_quality_gate_is_healthy(self) -> None:
outcomes = [_build_win_outcome(signal_id=index) for index in range(100)]
@@ -177,7 +186,7 @@ class RunBacktestScriptTests(unittest.TestCase):
self.assertEqual(exit_code, 2)
self.assertFalse(payload["passed"])
self.assertFalse(payload["quality_gate"]["passed"])
self.assertIn("sample_size_too_small", payload["quality_gate"]["reasons"])
self.assertIn("executed_sample_too_small", payload["quality_gate"]["reasons"])
def test_run_backtest_once_passes_replay_mode_and_config(self) -> None:
service = FakeBacktestService([])
@@ -201,6 +210,23 @@ class RunBacktestScriptTests(unittest.TestCase):
self.assertEqual(service.calls[0]["replay_config"].left_bars, 1)
self.assertEqual(service.calls[0]["replay_config"].right_bars, 1)
def test_run_backtest_once_passes_plan_replay_dimension(self) -> None:
service = FakeBacktestService([])
output = io.StringIO()
with contextlib.redirect_stdout(output):
exit_code = run_backtest_once(
service=service,
subject_type="plan_replay",
trading_plan_id=77,
candidate_model_code="OSOK",
)
self.assertEqual(exit_code, 0)
self.assertEqual(service.calls[0]["subject_type"], "plan_replay")
self.assertEqual(service.calls[0]["trading_plan_id"], 77)
self.assertEqual(service.calls[0]["candidate_model_code"], "OSOK")
def test_main_reports_database_unavailable(self) -> None:
output = io.StringIO()
tests/unit/test_backtest_service.py
+189
View File
@@ -6,6 +6,7 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:
from src.services.backtest import (
BacktestCostConfig,
BACKTEST_SUBJECT_PLAN_REPLAY,
build_timeframe_bucket,
classify_backtest_lifecycle_state,
BacktestOutcome,
@@ -17,6 +18,7 @@ from src.services.backtest import (
compute_max_consecutive_losses,
compute_max_drawdown_r,
compute_outcome_r_multiple,
build_plan_replay_units,
estimate_trade_cost_r,
evaluate_backtest_quality,
evaluate_signal_outcome,
@@ -378,6 +380,80 @@ class BacktestOutcomeTests(unittest.TestCase):
],
)
def test_plan_replay_quality_gate_uses_executed_sample_basis(self) -> None:
result = evaluate_backtest_quality(
{
"subject_type": BACKTEST_SUBJECT_PLAN_REPLAY,
"total_signals": 2,
"executed_signals": 35,
"expectancy_r": 0.2,
"profit_factor_r": 1.6,
"max_drawdown_r": 4.0,
"max_consecutive_losses": 2,
},
gate=BacktestQualityGate(min_total_signals=100, min_executed_signals=30),
sample_basis="executed",
)
self.assertTrue(result["passed"])
self.assertEqual(result["reasons"], [])
self.assertEqual(result["sample_basis"], "executed")
def test_build_plan_replay_units_tracks_appearance_and_confirmation_timeline(self) -> None:
rows = [
{
"candidate_id": 1,
"trading_plan_id": 77,
"candidate_model_code": "OSOK",
"side": "buy",
"setup_timeframe": "15m",
"trigger_timeframe": "5m",
"created_ts": datetime(2026, 5, 6, 0, 0, tzinfo=timezone.utc),
"expires_ts": datetime(2026, 5, 6, 4, 0, tzinfo=timezone.utc),
"confirmation_state": {"executable": False},
"missing_confirmations": ["sweep_confirmed"],
"entry": {},
"candidate_meta": {},
"candidate_status": "awaiting_confirmation",
},
{
"candidate_id": 2,
"trading_plan_id": 77,
"candidate_model_code": "OSOK",
"side": "buy",
"setup_timeframe": "15m",
"trigger_timeframe": "5m",
"created_ts": datetime(2026, 5, 6, 0, 5, tzinfo=timezone.utc),
"expires_ts": datetime(2026, 5, 6, 4, 0, tzinfo=timezone.utc),
"confirmation_state": {"executable": True},
"missing_confirmations": [],
"entry": {
"entry_low": 100.0,
"entry_high": 100.2,
"stop_loss": 99.0,
"tp1": 101.0,
"tp2": 103.0,
"target_plan": {
"status": "ready",
"tp1": {"price": 101.0, "objective_layer": "IRL", "source": {"type": "session_pivot", "table": "market_sessions", "field": "high"}},
"tp2": {"price": 103.0, "objective_layer": "ERL", "source": {"type": "dealing_range_boundary", "table": "dealing_ranges", "field": "high"}},
},
"management_rules": [{"code": "tp1_partial", "fraction": 0.5}],
},
"candidate_meta": {"management_rules": [{"code": "tp1_partial", "fraction": 0.5}]},
"candidate_status": "executable",
},
]
units = build_plan_replay_units(rows)
self.assertEqual(len(units), 1)
self.assertEqual(units[0]["candidate_id"], 2)
self.assertEqual(units[0]["appeared_ts"], rows[0]["created_ts"])
self.assertEqual(units[0]["confirmed_ts"], rows[1]["created_ts"])
self.assertAlmostEqual(units[0]["rr_tp1"], 0.8181818181)
self.assertEqual(units[0]["target_plan"]["tp2"]["objective_layer"], "ERL")
class BacktestServiceRunTests(unittest.TestCase):
def test_run_backtest_persists_quality_gate_in_summary(self) -> None:
@@ -530,6 +606,119 @@ class BacktestServiceRunTests(unittest.TestCase):
self.assertEqual(repository.results[0]["meta"]["backtest_mode"], "replay_no_lookahead")
self.assertEqual(repository.results[0]["meta"]["replay_confirmed_swing_count_total"], 2)
def test_run_backtest_supports_plan_replay_subject_and_plan_metrics(self) -> None:
case = self
class FakePlanReplayRepository:
def __init__(self) -> None:
self.summary = {}
self.results = []
self.run_config = {}
def fetch_active_trading_plan(self, instrument_id: int, trading_plan_id: int = None):
return {"id": 77}
def insert_backtest_run(self, **kwargs):
self.run_config = kwargs
return 501
def fetch_plan_replay_candidate_rows(self, **kwargs):
case.assertEqual(kwargs["trading_plan_id"], 77)
return [
{
"candidate_id": 11,
"trading_plan_id": 77,
"candidate_model_code": "OSOK",
"candidate_status": "awaiting_confirmation",
"side": "buy",
"setup_timeframe": "15m",
"trigger_timeframe": "5m",
"created_ts": datetime(2026, 5, 6, 0, 0, tzinfo=timezone.utc),
"expires_ts": datetime(2026, 5, 6, 4, 0, tzinfo=timezone.utc),
"confirmation_state": {"executable": False},
"missing_confirmations": ["sweep_confirmed"],
"entry": {},
"invalidations": [],
"candidate_meta": {},
"allowed_sessions": ["LONDON"],
"primary_draw": {"side": "buy_side"},
},
{
"candidate_id": 12,
"trading_plan_id": 77,
"candidate_model_code": "OSOK",
"candidate_status": "executable",
"side": "buy",
"setup_timeframe": "15m",
"trigger_timeframe": "5m",
"created_ts": datetime(2026, 5, 6, 0, 5, tzinfo=timezone.utc),
"expires_ts": datetime(2026, 5, 6, 4, 0, tzinfo=timezone.utc),
"confirmation_state": {"executable": True},
"missing_confirmations": [],
"entry": {
"entry_low": 100.0,
"entry_high": 100.2,
"stop_loss": 99.0,
"tp1": 101.0,
"tp2": 103.0,
"target_plan": {
"status": "ready",
"tp1": {"price": 101.0, "objective_layer": "IRL", "source": {"type": "session_pivot", "table": "market_sessions", "field": "high"}},
"tp2": {"price": 103.0, "objective_layer": "ERL", "source": {"type": "dealing_range_boundary", "table": "dealing_ranges", "field": "high"}},
},
"management_rules": [
{"code": "tp1_partial", "fraction": 0.5},
{"code": "move_stop", "to": "breakeven"},
],
},
"invalidations": [],
"candidate_meta": {
"management_rules": [
{"code": "tp1_partial", "fraction": 0.5},
{"code": "move_stop", "to": "breakeven"},
]
},
"allowed_sessions": ["LONDON"],
"primary_draw": {"side": "buy_side"},
},
]
def fetch_sessions_covering_window(self, **kwargs):
return [{"session_code": "LONDON"}]
def fetch_future_candles(self, **kwargs):
return [
{"ts_open": 1, "ts_close": 2, "high": 101.2, "low": 100.0, "open": 100.1, "close": 101.0},
{"ts_open": 2, "ts_close": 3, "high": 101.0, "low": 99.9, "open": 100.9, "close": 100.1},
]
def insert_backtest_result(self, **kwargs):
self.results.append(kwargs)
def update_backtest_run_summary(self, backtest_run_id: int, summary: dict) -> None:
self.summary = summary
repository = FakePlanReplayRepository()
service = BacktestService(repository=repository)
run_id, outcomes = service.run_backtest(
instrument_id=1,
subject_type=BACKTEST_SUBJECT_PLAN_REPLAY,
trading_plan_id=77,
timeframe_set={"bias_tf": "1h", "setup_tf": "15m", "trigger_tf": "5m"},
)
self.assertEqual(run_id, 501)
self.assertEqual(len(outcomes), 1)
self.assertEqual(repository.run_config["subject_type"], BACKTEST_SUBJECT_PLAN_REPLAY)
self.assertEqual(repository.results[0]["trading_plan_id"], 77)
self.assertEqual(repository.results[0]["candidate_id"], 12)
self.assertEqual(repository.summary["subject_type"], BACKTEST_SUBJECT_PLAN_REPLAY)
self.assertEqual(repository.summary["plan_count"], 1)
self.assertEqual(repository.summary["confirmed_candidates"], 1)
self.assertAlmostEqual(repository.summary["candidate_confirmation_rate"], 1.0)
self.assertIn("draw_target_hit_rate", repository.summary)
def test_normalize_timeframe_set_and_bucket_helpers(self) -> None:
normalized = normalize_timeframe_set({"bias_tf": "1h", "setup_tf": "15m"})