Codex
1741 lines
76 KiB
Python
1741 lines
76 KiB
Python
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
from src.repositories.backtest_repository import BacktestRepository
|
|
from src.services.backtest.replay import ReplaySwing, iter_historical_replay
|
|
from src.services.ict_mainline import (
|
|
build_doctrine_backtest_cohort,
|
|
build_liquidity_backtest_features,
|
|
build_pd_array_backtest_features,
|
|
build_time_session_backtest_features,
|
|
evaluate_liquidity_kernel_status,
|
|
evaluate_pd_array_reaction_status,
|
|
evaluate_time_session_kernel_status,
|
|
extract_setup_playbook_id,
|
|
)
|
|
|
|
|
|
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,
|
|
}
|
|
|
|
BACKTEST_LIFECYCLE_OPEN = "open"
|
|
BACKTEST_LIFECYCLE_REJECTED = "rejected"
|
|
BACKTEST_LIFECYCLE_AWAITING_CONFIRMATION = "awaiting_confirmation"
|
|
BACKTEST_LIFECYCLE_INVALIDATED_BEFORE_ENTRY = "invalidated_before_entry"
|
|
BACKTEST_LIFECYCLE_MISSED = "missed"
|
|
BACKTEST_LIFECYCLE_STOPPED_OUT = "stopped_out"
|
|
BACKTEST_LIFECYCLE_TP1_REACHED = "tp1_reached"
|
|
BACKTEST_LIFECYCLE_TP2_CLOSED = "tp2_closed"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BacktestOutcome:
|
|
signal_id: Optional[int]
|
|
outcome: str
|
|
mfe: float
|
|
mae: float
|
|
tp1_hit: bool
|
|
tp2_hit: bool
|
|
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)
|
|
class BacktestSignal:
|
|
signal_id: Optional[int]
|
|
side: str
|
|
entry_low: float
|
|
entry_high: float
|
|
stop_loss: float
|
|
tp1: float
|
|
tp2: float
|
|
rr_tp1: float = 0.0
|
|
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)
|
|
class BacktestCostConfig:
|
|
fee_rate: float = 0.0
|
|
slippage_rate: float = 0.0
|
|
fixed_cost_r: float = 0.0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BacktestReplayConfig:
|
|
left_bars: int = 2
|
|
right_bars: int = 2
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BacktestQualityGate:
|
|
min_total_signals: int = 100
|
|
min_executed_signals: int = 30
|
|
min_expectancy_r: float = 0.0
|
|
min_profit_factor_r: float = 1.2
|
|
max_drawdown_r: float = 10.0
|
|
max_consecutive_losses: int = 5
|
|
|
|
|
|
class BacktestService:
|
|
def __init__(self, repository: Optional[BacktestRepository] = None) -> None:
|
|
self.repository = repository or BacktestRepository()
|
|
|
|
def run_backtest(
|
|
self,
|
|
instrument_id: int,
|
|
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,
|
|
session_codes: Optional[list[str]] = None,
|
|
run_metadata: Optional[dict] = None,
|
|
) -> tuple[int, list[BacktestOutcome]]:
|
|
cost_config = cost_config or BacktestCostConfig()
|
|
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}")
|
|
active_trading_plan = None
|
|
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_metadata = dict(run_metadata or {})
|
|
setup_playbook_id = extract_setup_playbook_id(run_metadata)
|
|
if not setup_playbook_id and active_trading_plan:
|
|
setup_playbook_id = extract_setup_playbook_id(active_trading_plan)
|
|
if setup_playbook_id:
|
|
run_metadata['setup_playbook_id'] = setup_playbook_id
|
|
liquidity_backtest_features = build_liquidity_backtest_features(
|
|
{
|
|
'subject_type': subject_type,
|
|
'model_code': 'SWEEP_MSS_FVG',
|
|
'candidate_model_code': candidate_model_code,
|
|
'timeframe_set': timeframe_set,
|
|
'run_metadata': run_metadata,
|
|
'active_trading_plan': active_trading_plan,
|
|
}
|
|
)
|
|
pd_array_backtest_features = build_pd_array_backtest_features(
|
|
{
|
|
'subject_type': subject_type,
|
|
'model_code': 'SWEEP_MSS_FVG',
|
|
'candidate_model_code': candidate_model_code,
|
|
'timeframe_set': timeframe_set,
|
|
'run_metadata': run_metadata,
|
|
'active_trading_plan': active_trading_plan,
|
|
}
|
|
)
|
|
time_session_backtest_features = build_time_session_backtest_features(
|
|
{
|
|
'subject_type': subject_type,
|
|
'model_code': 'SWEEP_MSS_FVG',
|
|
'candidate_model_code': candidate_model_code,
|
|
'timeframe_set': timeframe_set,
|
|
'session_codes': list(session_codes or []),
|
|
'run_metadata': run_metadata,
|
|
'active_trading_plan': active_trading_plan,
|
|
}
|
|
)
|
|
doctrine_backtest_cohort = build_doctrine_backtest_cohort(
|
|
{
|
|
'subject_type': subject_type,
|
|
'model_code': 'SWEEP_MSS_FVG',
|
|
'candidate_model_code': candidate_model_code,
|
|
'timeframe_set': timeframe_set,
|
|
'run_metadata': run_metadata,
|
|
'liquidity_backtest_features': liquidity_backtest_features,
|
|
'pd_array_backtest_features': pd_array_backtest_features,
|
|
'time_session_backtest_features': time_session_backtest_features,
|
|
}
|
|
)
|
|
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__,
|
|
'quality_gate': quality_gate.__dict__,
|
|
'replay': replay_config.__dict__ if mode == BACKTEST_MODE_REPLAY_NO_LOOKAHEAD else None,
|
|
'session_codes': list(session_codes or []),
|
|
'setup_playbook_id': setup_playbook_id,
|
|
'liquidity_backtest_features': liquidity_backtest_features,
|
|
'liquidity_kernel_status': evaluate_liquidity_kernel_status(liquidity_backtest_features),
|
|
'pd_array_backtest_features': pd_array_backtest_features,
|
|
'pd_array_reaction_status': evaluate_pd_array_reaction_status(pd_array_backtest_features),
|
|
'time_session_backtest_features': time_session_backtest_features,
|
|
'time_session_kernel_status': evaluate_time_session_kernel_status(time_session_backtest_features),
|
|
'doctrine_backtest_cohort': doctrine_backtest_cohort,
|
|
'run_metadata': run_metadata,
|
|
},
|
|
)
|
|
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')
|
|
outcome_setup_playbook_id = setup_playbook_id or _first_outcome_meta_value(outcomes, 'setup_playbook_id')
|
|
outcome_liquidity_features = _first_outcome_meta_value(outcomes, 'liquidity_backtest_features') or liquidity_backtest_features
|
|
outcome_pd_array_features = _first_outcome_meta_value(outcomes, 'pd_array_backtest_features') or pd_array_backtest_features
|
|
outcome_time_session_features = _first_outcome_meta_value(outcomes, 'time_session_backtest_features') or time_session_backtest_features
|
|
summary['liquidity_backtest_features'] = dict(outcome_liquidity_features or {})
|
|
summary['liquidity_kernel_status'] = evaluate_liquidity_kernel_status(summary['liquidity_backtest_features'])
|
|
summary['pd_array_backtest_features'] = dict(outcome_pd_array_features or {})
|
|
summary['pd_array_reaction_status'] = evaluate_pd_array_reaction_status(summary['pd_array_backtest_features'])
|
|
summary['time_session_backtest_features'] = dict(outcome_time_session_features or {})
|
|
summary['time_session_kernel_status'] = evaluate_time_session_kernel_status(summary['time_session_backtest_features'])
|
|
summary['doctrine_backtest_cohort'] = build_doctrine_backtest_cohort(
|
|
{
|
|
'subject_type': subject_type,
|
|
'model_code': 'SWEEP_MSS_FVG',
|
|
'candidate_model_code': candidate_model_code,
|
|
'timeframe_set': timeframe_set,
|
|
'run_metadata': {**run_metadata, 'setup_playbook_id': outcome_setup_playbook_id},
|
|
'summary': summary,
|
|
'liquidity_backtest_features': summary['liquidity_backtest_features'],
|
|
'pd_array_backtest_features': summary['pd_array_backtest_features'],
|
|
'time_session_backtest_features': summary['time_session_backtest_features'],
|
|
},
|
|
quality_gate=summary['quality_gate'],
|
|
)
|
|
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,
|
|
session_codes=session_codes,
|
|
)
|
|
outcomes: list[BacktestOutcome] = []
|
|
for signal in signals:
|
|
sessions = self.repository.fetch_sessions_covering_window(
|
|
instrument_id=instrument_id,
|
|
start_ts=signal['created_ts'],
|
|
end_ts=signal['created_ts'],
|
|
)
|
|
future = self.repository.fetch_future_candles(
|
|
instrument_id=instrument_id,
|
|
timeframe=str(signal.get('trigger_tf') or timeframe_set['trigger_tf']),
|
|
start_ts=signal['created_ts'],
|
|
end_ts=signal['expires_ts'],
|
|
limit=500,
|
|
)
|
|
if not future:
|
|
target_plan = extract_target_plan_from_signal_row(signal)
|
|
management_rules = extract_management_rules_from_signal_row(signal)
|
|
session_codes = resolve_signal_session_codes(signal=signal, sessions=sessions)
|
|
signal_liquidity_features = build_liquidity_backtest_features(
|
|
{
|
|
'signal': signal,
|
|
'target_plan': target_plan,
|
|
'run_metadata': run_metadata,
|
|
}
|
|
)
|
|
signal_pd_array_features = build_pd_array_backtest_features(
|
|
{
|
|
'signal': signal,
|
|
'target_plan': target_plan,
|
|
'run_metadata': run_metadata,
|
|
}
|
|
)
|
|
signal_time_session_features = build_time_session_backtest_features(
|
|
{
|
|
'signal': signal,
|
|
'target_plan': target_plan,
|
|
'run_metadata': run_metadata,
|
|
'session_codes': session_codes,
|
|
'session_filter': 'outside_window' if len(sessions) == 0 else 'covered',
|
|
}
|
|
)
|
|
error_tags = ['no_future_candles']
|
|
if len(sessions) == 0:
|
|
error_tags.append('outside_session_window')
|
|
record = BacktestOutcome(
|
|
signal_id=signal['id'],
|
|
outcome='missed',
|
|
mfe=0.0,
|
|
mae=0.0,
|
|
tp1_hit=False,
|
|
tp2_hit=False,
|
|
invalidated_before_entry=True,
|
|
error_tags=_dedupe_text_list(error_tags),
|
|
meta={
|
|
'entry_triggered': False,
|
|
'same_candle_priority': SAME_CANDLE_PRIORITY,
|
|
'same_candle_ambiguous': False,
|
|
'sessions_found': len(sessions),
|
|
'session_codes': session_codes,
|
|
'session_filter': 'outside_window' if len(sessions) == 0 else 'covered',
|
|
'signal_side': signal['side'],
|
|
'signal_symbol': signal.get('symbol'),
|
|
'signal_timeframe_set': {
|
|
'bias_tf': signal.get('bias_tf') or timeframe_set['bias_tf'],
|
|
'setup_tf': signal.get('setup_tf') or timeframe_set['setup_tf'],
|
|
'trigger_tf': signal.get('trigger_tf') or timeframe_set['trigger_tf'],
|
|
},
|
|
'signal_timeframe_bucket': build_timeframe_bucket(
|
|
bias_tf=signal.get('bias_tf') or timeframe_set['bias_tf'],
|
|
setup_tf=signal.get('setup_tf') or timeframe_set['setup_tf'],
|
|
trigger_tf=signal.get('trigger_tf') or timeframe_set['trigger_tf'],
|
|
),
|
|
'signal_created_ts': signal['created_ts'],
|
|
'rr_tp1': float(signal.get('rr_tp1') or 0.0),
|
|
'rr_tp2': float(signal.get('rr_tp2') or 0.0),
|
|
'target_plan': target_plan,
|
|
'management_rules': management_rules,
|
|
'liquidity_backtest_features': signal_liquidity_features,
|
|
'liquidity_kernel_status': evaluate_liquidity_kernel_status(signal_liquidity_features),
|
|
'pd_array_backtest_features': signal_pd_array_features,
|
|
'pd_array_reaction_status': evaluate_pd_array_reaction_status(signal_pd_array_features),
|
|
'time_session_backtest_features': signal_time_session_features,
|
|
'time_session_kernel_status': evaluate_time_session_kernel_status(signal_time_session_features),
|
|
'cost_r': 0.0,
|
|
'cost_model': cost_config.__dict__,
|
|
'backtest_mode': mode,
|
|
'r_multiple': 0.0,
|
|
'weekday': _extract_weekday(signal['created_ts']),
|
|
'replay_left_bars': replay_config.left_bars if mode == BACKTEST_MODE_REPLAY_NO_LOOKAHEAD else None,
|
|
'replay_right_bars': replay_config.right_bars if mode == BACKTEST_MODE_REPLAY_NO_LOOKAHEAD else None,
|
|
},
|
|
)
|
|
|
|
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,
|
|
)
|
|
continue
|
|
|
|
session_filtered = len(sessions) == 0
|
|
session_codes = resolve_signal_session_codes(signal=signal, sessions=sessions)
|
|
record = evaluate_signal_outcome(
|
|
BacktestSignal(
|
|
signal_id=signal['id'],
|
|
side=signal['side'],
|
|
entry_low=float(signal['entry_low'] or 0),
|
|
entry_high=float(signal['entry_high'] or 0),
|
|
stop_loss=float(signal['stop_loss'] or 0),
|
|
tp1=float(signal['tp1'] or 0),
|
|
tp2=float(signal['tp2'] or 0),
|
|
rr_tp1=float(signal.get('rr_tp1') or 0),
|
|
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,
|
|
mode=mode,
|
|
replay_config=replay_config,
|
|
)
|
|
if session_filtered:
|
|
record.error_tags.append('outside_session_window')
|
|
record.meta.update(
|
|
{
|
|
'sessions_found': len(sessions),
|
|
'session_codes': session_codes,
|
|
'session_filter': 'outside_window' if session_filtered else 'covered',
|
|
'signal_side': signal['side'],
|
|
'signal_symbol': signal.get('symbol'),
|
|
'signal_timeframe_set': {
|
|
'bias_tf': signal.get('bias_tf') or timeframe_set['bias_tf'],
|
|
'setup_tf': signal.get('setup_tf') or timeframe_set['setup_tf'],
|
|
'trigger_tf': signal.get('trigger_tf') or timeframe_set['trigger_tf'],
|
|
},
|
|
'signal_timeframe_bucket': build_timeframe_bucket(
|
|
bias_tf=signal.get('bias_tf') or timeframe_set['bias_tf'],
|
|
setup_tf=signal.get('setup_tf') or timeframe_set['setup_tf'],
|
|
trigger_tf=signal.get('trigger_tf') or timeframe_set['trigger_tf'],
|
|
),
|
|
'signal_created_ts': signal['created_ts'],
|
|
'rr_tp1': float(signal.get('rr_tp1') or 0.0),
|
|
'rr_tp2': float(signal.get('rr_tp2') or 0.0),
|
|
'target_plan': extract_target_plan_from_signal_row(signal),
|
|
'management_rules': extract_management_rules_from_signal_row(signal),
|
|
'liquidity_backtest_features': build_liquidity_backtest_features(
|
|
{
|
|
'signal': signal,
|
|
'target_plan': extract_target_plan_from_signal_row(signal),
|
|
'run_metadata': run_metadata,
|
|
}
|
|
),
|
|
'pd_array_backtest_features': build_pd_array_backtest_features(
|
|
{
|
|
'signal': signal,
|
|
'target_plan': extract_target_plan_from_signal_row(signal),
|
|
'run_metadata': run_metadata,
|
|
}
|
|
),
|
|
'time_session_backtest_features': build_time_session_backtest_features(
|
|
{
|
|
'signal': signal,
|
|
'target_plan': extract_target_plan_from_signal_row(signal),
|
|
'run_metadata': run_metadata,
|
|
'session_codes': session_codes,
|
|
'session_filter': 'outside_window' if session_filtered else 'covered',
|
|
}
|
|
),
|
|
'weekday': _extract_weekday(signal['created_ts']),
|
|
}
|
|
)
|
|
record.meta['liquidity_kernel_status'] = evaluate_liquidity_kernel_status(record.meta['liquidity_backtest_features'])
|
|
record.meta['pd_array_reaction_status'] = evaluate_pd_array_reaction_status(record.meta['pd_array_backtest_features'])
|
|
record.meta['time_session_kernel_status'] = evaluate_time_session_kernel_status(record.meta['time_session_backtest_features'])
|
|
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,
|
|
)
|
|
summary = summarize_backtest_outcomes(outcomes, subject_type=subject_type)
|
|
summary['quality_gate'] = evaluate_backtest_quality(summary, gate=quality_gate)
|
|
outcome_liquidity_features = _first_outcome_meta_value(outcomes, 'liquidity_backtest_features') or liquidity_backtest_features
|
|
outcome_pd_array_features = _first_outcome_meta_value(outcomes, 'pd_array_backtest_features') or pd_array_backtest_features
|
|
outcome_time_session_features = _first_outcome_meta_value(outcomes, 'time_session_backtest_features') or time_session_backtest_features
|
|
summary['liquidity_backtest_features'] = dict(outcome_liquidity_features or {})
|
|
summary['liquidity_kernel_status'] = evaluate_liquidity_kernel_status(summary['liquidity_backtest_features'])
|
|
summary['pd_array_backtest_features'] = dict(outcome_pd_array_features or {})
|
|
summary['pd_array_reaction_status'] = evaluate_pd_array_reaction_status(summary['pd_array_backtest_features'])
|
|
summary['time_session_backtest_features'] = dict(outcome_time_session_features or {})
|
|
summary['time_session_kernel_status'] = evaluate_time_session_kernel_status(summary['time_session_backtest_features'])
|
|
summary['doctrine_backtest_cohort'] = build_doctrine_backtest_cohort(
|
|
{
|
|
'subject_type': subject_type,
|
|
'model_code': 'SWEEP_MSS_FVG',
|
|
'candidate_model_code': candidate_model_code,
|
|
'timeframe_set': timeframe_set,
|
|
'run_metadata': run_metadata,
|
|
'summary': summary,
|
|
'liquidity_backtest_features': summary['liquidity_backtest_features'],
|
|
'pd_array_backtest_features': summary['pd_array_backtest_features'],
|
|
'time_session_backtest_features': summary['time_session_backtest_features'],
|
|
},
|
|
quality_gate=summary['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:
|
|
session_reference_ts = unit['confirmed_ts'] or unit['appeared_ts']
|
|
sessions = self.repository.fetch_sessions_covering_window(
|
|
instrument_id=instrument_id,
|
|
start_ts=session_reference_ts,
|
|
end_ts=session_reference_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=None,
|
|
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=None,
|
|
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=None,
|
|
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=None,
|
|
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,
|
|
future_candles: list[dict],
|
|
cost_config: Optional[BacktestCostConfig] = None,
|
|
mode: str = BACKTEST_MODE_ENTRY_THEN_OUTCOME,
|
|
replay_config: Optional[BacktestReplayConfig] = None,
|
|
) -> BacktestOutcome:
|
|
if mode == BACKTEST_MODE_REPLAY_NO_LOOKAHEAD:
|
|
return evaluate_signal_outcome_with_replay(
|
|
signal=signal,
|
|
future_candles=future_candles,
|
|
cost_config=cost_config,
|
|
replay_config=replay_config,
|
|
)
|
|
cost_config = cost_config or BacktestCostConfig()
|
|
steps = [
|
|
_BacktestEvaluationStep(
|
|
index=index,
|
|
candle=candle,
|
|
ts=_extract_candle_ts(candle),
|
|
known_candles_count=index + 1,
|
|
confirmed_swings_available=0,
|
|
last_confirmed_swing=None,
|
|
)
|
|
for index, candle in enumerate(future_candles)
|
|
]
|
|
return _evaluate_signal_outcome_steps(
|
|
signal=signal,
|
|
steps=steps,
|
|
cost_config=cost_config,
|
|
mode=BACKTEST_MODE_ENTRY_THEN_OUTCOME,
|
|
replay_config=None,
|
|
)
|
|
|
|
|
|
def evaluate_signal_outcome_with_replay(
|
|
signal: BacktestSignal,
|
|
future_candles: list[dict],
|
|
cost_config: Optional[BacktestCostConfig] = None,
|
|
replay_config: Optional[BacktestReplayConfig] = None,
|
|
) -> BacktestOutcome:
|
|
cost_config = cost_config or BacktestCostConfig()
|
|
replay_config = replay_config or BacktestReplayConfig()
|
|
try:
|
|
steps = _build_replay_evaluation_steps(future_candles=future_candles, replay_config=replay_config)
|
|
except ValueError as exc:
|
|
return BacktestOutcome(
|
|
signal_id=signal.signal_id,
|
|
outcome='missed',
|
|
mfe=0.0,
|
|
mae=0.0,
|
|
tp1_hit=False,
|
|
tp2_hit=False,
|
|
invalidated_before_entry=True,
|
|
error_tags=['invalid_replay_candles'],
|
|
meta={
|
|
'entry_triggered': False,
|
|
'same_candle_priority': SAME_CANDLE_PRIORITY,
|
|
'same_candle_ambiguous': False,
|
|
'signal_side': signal.side,
|
|
'rr_tp1': signal.rr_tp1,
|
|
'rr_tp2': signal.rr_tp2,
|
|
'mfe_r': 0.0,
|
|
'mae_r': 0.0,
|
|
'cost_r': 0.0,
|
|
'cost_model': cost_config.__dict__,
|
|
'backtest_mode': BACKTEST_MODE_REPLAY_NO_LOOKAHEAD,
|
|
'r_multiple': 0.0,
|
|
'replay_error': str(exc),
|
|
'replay_frame_count': 0,
|
|
'replay_entry_frame_index': None,
|
|
'replay_exit_frame_index': None,
|
|
'replay_known_candles_at_entry': 0,
|
|
'replay_confirmed_swing_count_at_entry': 0,
|
|
'replay_last_confirmed_swing_at_entry': None,
|
|
'replay_confirmed_swing_count_total': 0,
|
|
'replay_left_bars': replay_config.left_bars,
|
|
'replay_right_bars': replay_config.right_bars,
|
|
},
|
|
)
|
|
return _evaluate_signal_outcome_steps(
|
|
signal=signal,
|
|
steps=steps,
|
|
cost_config=cost_config,
|
|
mode=BACKTEST_MODE_REPLAY_NO_LOOKAHEAD,
|
|
replay_config=replay_config,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _BacktestEvaluationStep:
|
|
index: int
|
|
candle: dict
|
|
ts: object
|
|
known_candles_count: int
|
|
confirmed_swings_available: int
|
|
last_confirmed_swing: Optional[dict]
|
|
|
|
|
|
def _build_replay_evaluation_steps(
|
|
future_candles: list[dict],
|
|
replay_config: BacktestReplayConfig,
|
|
) -> list[_BacktestEvaluationStep]:
|
|
steps: list[_BacktestEvaluationStep] = []
|
|
confirmed_swings: list[ReplaySwing] = []
|
|
for frame in iter_historical_replay(
|
|
future_candles,
|
|
left_bars=replay_config.left_bars,
|
|
right_bars=replay_config.right_bars,
|
|
):
|
|
confirmed_swings.extend(frame.new_swings)
|
|
steps.append(
|
|
_BacktestEvaluationStep(
|
|
index=frame.index,
|
|
candle=frame.candle,
|
|
ts=_extract_candle_ts(frame.candle),
|
|
known_candles_count=len(frame.known_candles),
|
|
confirmed_swings_available=len(confirmed_swings),
|
|
last_confirmed_swing=_serialize_replay_swing(confirmed_swings[-1]) if confirmed_swings else None,
|
|
)
|
|
)
|
|
return steps
|
|
|
|
|
|
def _evaluate_signal_outcome_steps(
|
|
signal: BacktestSignal,
|
|
steps: list[_BacktestEvaluationStep],
|
|
cost_config: BacktestCostConfig,
|
|
mode: str,
|
|
replay_config: Optional[BacktestReplayConfig],
|
|
) -> BacktestOutcome:
|
|
entry_mid = (signal.entry_low + signal.entry_high) / 2
|
|
management_rules = list(signal.management_rules or [])
|
|
tp1_partial_fraction = resolve_tp1_partial_fraction(management_rules)
|
|
move_stop_enabled = has_management_rule(management_rules, "move_stop")
|
|
entry_triggered = False
|
|
tp1_hit = False
|
|
tp2_hit = False
|
|
stopped_out = False
|
|
stopped_after_tp1 = False
|
|
same_candle_ambiguous = False
|
|
mfe = 0.0
|
|
mae = 0.0
|
|
entry_step: Optional[_BacktestEvaluationStep] = None
|
|
exit_step: Optional[_BacktestEvaluationStep] = None
|
|
active_stop_loss = signal.stop_loss
|
|
|
|
for step in steps:
|
|
candle = step.candle
|
|
high = float(candle['high'])
|
|
low = float(candle['low'])
|
|
entry_now = low <= signal.entry_high and high >= signal.entry_low
|
|
entry_triggered_this_candle = False
|
|
|
|
if not entry_triggered:
|
|
if entry_now:
|
|
entry_triggered = True
|
|
entry_triggered_this_candle = True
|
|
entry_step = step
|
|
else:
|
|
continue
|
|
|
|
if signal.side == 'buy':
|
|
mfe = max(mfe, high - entry_mid)
|
|
mae = max(mae, entry_mid - low)
|
|
stop_hit = bool(active_stop_loss and low <= active_stop_loss)
|
|
tp1_now = bool(signal.tp1 and high >= signal.tp1)
|
|
tp2_now = bool(signal.tp2 and high >= signal.tp2)
|
|
else:
|
|
mfe = max(mfe, entry_mid - low)
|
|
mae = max(mae, high - entry_mid)
|
|
stop_hit = bool(active_stop_loss and high >= active_stop_loss)
|
|
tp1_now = bool(signal.tp1 and low <= signal.tp1)
|
|
tp2_now = bool(signal.tp2 and low <= signal.tp2)
|
|
|
|
if entry_triggered_this_candle and (stop_hit or tp1_now or tp2_now):
|
|
same_candle_ambiguous = True
|
|
if stop_hit and (tp1_now or tp2_now):
|
|
same_candle_ambiguous = True
|
|
|
|
if stop_hit:
|
|
stopped_out = True
|
|
stopped_after_tp1 = tp1_hit
|
|
exit_step = step
|
|
break
|
|
if tp1_now:
|
|
tp1_hit = True
|
|
if move_stop_enabled:
|
|
active_stop_loss = entry_mid
|
|
if tp2_now:
|
|
tp2_hit = True
|
|
exit_step = step
|
|
break
|
|
|
|
error_tags: list[str] = []
|
|
if same_candle_ambiguous:
|
|
error_tags.append('same_candle_entry_exit_ambiguous')
|
|
|
|
if not entry_triggered:
|
|
outcome = 'missed'
|
|
error_tags.append('entry_not_triggered')
|
|
elif tp2_hit:
|
|
outcome = 'win'
|
|
elif stopped_out and tp1_hit and move_stop_enabled:
|
|
outcome = 'breakeven'
|
|
elif tp1_hit and not stopped_out:
|
|
outcome = 'breakeven'
|
|
elif stopped_out:
|
|
outcome = 'loss'
|
|
error_tags.append('stopped_out')
|
|
else:
|
|
outcome = 'missed'
|
|
error_tags.append('no_target_hit')
|
|
risk_distance = abs(entry_mid - signal.stop_loss)
|
|
mfe_r = mfe / risk_distance if risk_distance else 0.0
|
|
mae_r = mae / risk_distance if risk_distance else 0.0
|
|
cost_r = estimate_trade_cost_r(
|
|
signal=signal,
|
|
outcome=outcome,
|
|
entry_price=entry_mid,
|
|
entry_triggered=entry_triggered,
|
|
cost_config=cost_config,
|
|
)
|
|
r_multiple = compute_outcome_r_multiple(
|
|
outcome=outcome,
|
|
rr_tp1=signal.rr_tp1,
|
|
rr_tp2=signal.rr_tp2,
|
|
cost_r=cost_r,
|
|
tp1_hit=tp1_hit,
|
|
tp2_hit=tp2_hit,
|
|
tp1_partial_fraction=tp1_partial_fraction,
|
|
stopped_after_tp1=stopped_after_tp1,
|
|
move_stop_applied=move_stop_enabled,
|
|
)
|
|
remaining_position_fraction = max(0.0, 1.0 - tp1_partial_fraction) if tp1_hit else 1.0
|
|
management_applied = {
|
|
"tp1_partial": tp1_partial_fraction > 0 and tp1_hit,
|
|
"tp1_partial_fraction": tp1_partial_fraction if tp1_hit else 0.0,
|
|
"move_stop": move_stop_enabled and tp1_hit,
|
|
"move_stop_to": "breakeven" if move_stop_enabled and tp1_hit else None,
|
|
"remaining_position_fraction": remaining_position_fraction,
|
|
"remaining_position_closed": bool(tp2_hit or (stopped_after_tp1 and move_stop_enabled)),
|
|
"tp1_realized_r": (tp1_partial_fraction * signal.rr_tp1) if tp1_hit else 0.0,
|
|
"active_stop_loss_after_tp1": entry_mid if move_stop_enabled and tp1_hit else signal.stop_loss,
|
|
}
|
|
|
|
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,
|
|
'rr_tp1': signal.rr_tp1,
|
|
'rr_tp2': signal.rr_tp2,
|
|
'target_plan': dict(signal.target_plan or {}),
|
|
'management_rules': list(signal.management_rules or []),
|
|
'mfe_r': mfe_r,
|
|
'mae_r': mae_r,
|
|
'cost_r': cost_r,
|
|
'cost_model': cost_config.__dict__,
|
|
'backtest_mode': mode,
|
|
'r_multiple': r_multiple,
|
|
'management_applied': management_applied,
|
|
}
|
|
if mode == BACKTEST_MODE_REPLAY_NO_LOOKAHEAD:
|
|
meta.update(
|
|
{
|
|
'replay_frame_count': len(steps),
|
|
'replay_entry_frame_index': entry_step.index if entry_step else None,
|
|
'replay_exit_frame_index': exit_step.index if exit_step else None,
|
|
'replay_entry_ts': _normalize_meta_ts(entry_step.ts) if entry_step else None,
|
|
'replay_exit_ts': _normalize_meta_ts(exit_step.ts) if exit_step else None,
|
|
'replay_known_candles_at_entry': entry_step.known_candles_count if entry_step else 0,
|
|
'replay_confirmed_swing_count_at_entry': entry_step.confirmed_swings_available if entry_step else 0,
|
|
'replay_last_confirmed_swing_at_entry': entry_step.last_confirmed_swing if entry_step else None,
|
|
'replay_confirmed_swing_count_total': steps[-1].confirmed_swings_available if steps else 0,
|
|
'replay_left_bars': replay_config.left_bars if replay_config else None,
|
|
'replay_right_bars': replay_config.right_bars if replay_config else None,
|
|
}
|
|
)
|
|
|
|
return BacktestOutcome(
|
|
signal_id=signal.signal_id,
|
|
outcome=outcome,
|
|
mfe=mfe,
|
|
mae=mae,
|
|
tp1_hit=tp1_hit,
|
|
tp2_hit=tp2_hit,
|
|
invalidated_before_entry=False,
|
|
error_tags=error_tags,
|
|
meta=meta,
|
|
)
|
|
|
|
|
|
def _serialize_replay_swing(swing: ReplaySwing) -> dict:
|
|
return {
|
|
'kind': swing.kind,
|
|
'price': swing.price,
|
|
'ts': _normalize_meta_ts(swing.ts),
|
|
'source_index': swing.source_index,
|
|
'confirmed_at_index': swing.confirmed_at_index,
|
|
'left_bars': swing.left_bars,
|
|
'right_bars': swing.right_bars,
|
|
}
|
|
|
|
|
|
def _extract_candle_ts(candle: dict):
|
|
return candle.get('ts_open') or candle.get('ts_close')
|
|
|
|
|
|
def _normalize_meta_ts(value):
|
|
if hasattr(value, 'isoformat'):
|
|
return value.isoformat()
|
|
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 {})
|
|
management_rules = list((confirmed_row or latest).get('management_rules') or target_plan.get('management_rules') or [])
|
|
setup_playbook_id = (
|
|
extract_setup_playbook_id(first)
|
|
or extract_setup_playbook_id(latest)
|
|
or extract_setup_playbook_id(confirmed_row or {})
|
|
)
|
|
doctrine_backtest_cohort = build_doctrine_backtest_cohort(
|
|
{
|
|
'setup_playbook_id': setup_playbook_id,
|
|
'candidate_model_code': first.get('candidate_model_code'),
|
|
'timeframe_set': {
|
|
'setup_tf': first.get('setup_timeframe'),
|
|
'trigger_tf': first.get('trigger_timeframe'),
|
|
},
|
|
}
|
|
)
|
|
liquidity_backtest_features = build_liquidity_backtest_features(
|
|
{
|
|
'first_candidate': first,
|
|
'latest_candidate': latest,
|
|
'confirmed_candidate': confirmed_row or {},
|
|
'entry_payload': entry_payload,
|
|
'target_plan': target_plan,
|
|
'setup_playbook_id': setup_playbook_id,
|
|
}
|
|
)
|
|
liquidity_kernel_status = evaluate_liquidity_kernel_status(liquidity_backtest_features)
|
|
pd_array_backtest_features = build_pd_array_backtest_features(
|
|
{
|
|
'first_candidate': first,
|
|
'latest_candidate': latest,
|
|
'confirmed_candidate': confirmed_row or {},
|
|
'entry_payload': entry_payload,
|
|
'target_plan': target_plan,
|
|
'setup_playbook_id': setup_playbook_id,
|
|
}
|
|
)
|
|
pd_array_reaction_status = evaluate_pd_array_reaction_status(pd_array_backtest_features)
|
|
time_session_backtest_features = build_time_session_backtest_features(
|
|
{
|
|
'first_candidate': first,
|
|
'latest_candidate': latest,
|
|
'confirmed_candidate': confirmed_row or {},
|
|
'entry_payload': entry_payload,
|
|
'target_plan': target_plan,
|
|
'setup_playbook_id': setup_playbook_id,
|
|
'session_code': first.get('session_code'),
|
|
'allowed_sessions': list(first.get('allowed_sessions') or []),
|
|
}
|
|
)
|
|
time_session_kernel_status = evaluate_time_session_kernel_status(time_session_backtest_features)
|
|
doctrine_backtest_cohort = build_doctrine_backtest_cohort(
|
|
{
|
|
'setup_playbook_id': setup_playbook_id,
|
|
'candidate_model_code': first.get('candidate_model_code'),
|
|
'timeframe_set': {
|
|
'setup_tf': first.get('setup_timeframe'),
|
|
'trigger_tf': first.get('trigger_timeframe'),
|
|
},
|
|
'liquidity_backtest_features': liquidity_backtest_features,
|
|
'pd_array_backtest_features': pd_array_backtest_features,
|
|
'time_session_backtest_features': time_session_backtest_features,
|
|
}
|
|
)
|
|
units.append(
|
|
{
|
|
'trading_plan_id': first.get('trading_plan_id'),
|
|
'candidate_id': (confirmed_row or first).get('candidate_id'),
|
|
'setup_playbook_id': setup_playbook_id,
|
|
'doctrine_backtest_cohort': doctrine_backtest_cohort,
|
|
'liquidity_backtest_features': liquidity_backtest_features,
|
|
'liquidity_kernel_status': liquidity_kernel_status,
|
|
'pd_array_backtest_features': pd_array_backtest_features,
|
|
'pd_array_reaction_status': pd_array_reaction_status,
|
|
'time_session_backtest_features': time_session_backtest_features,
|
|
'time_session_kernel_status': time_session_kernel_status,
|
|
'candidate_model_code': first.get('candidate_model_code'),
|
|
'side': first.get('side'),
|
|
'session_code': first.get('session_code'),
|
|
'candidate_state': latest.get('candidate_state'),
|
|
'quality_score': latest.get('quality_score'),
|
|
'ai_explanation': latest.get('ai_explanation'),
|
|
'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('expiry_ts') or (confirmed_row or latest).get('expires_ts') or latest.get('created_ts'),
|
|
'entry_low': (confirmed_row or latest).get('entry_low'),
|
|
'entry_high': (confirmed_row or latest).get('entry_high'),
|
|
'stop_loss': (confirmed_row or latest).get('stop_loss'),
|
|
'tp1': (confirmed_row or latest).get('tp1'),
|
|
'tp2': (confirmed_row or latest).get('tp2'),
|
|
'rr_tp1': derive_rr_from_entry(
|
|
side=first.get('side'),
|
|
entry_low=(confirmed_row or latest).get('entry_low'),
|
|
entry_high=(confirmed_row or latest).get('entry_high'),
|
|
stop_loss=(confirmed_row or latest).get('stop_loss'),
|
|
target_price=(confirmed_row or latest).get('tp1'),
|
|
),
|
|
'rr_tp2': derive_rr_from_entry(
|
|
side=first.get('side'),
|
|
entry_low=(confirmed_row or latest).get('entry_low'),
|
|
entry_high=(confirmed_row or latest).get('entry_high'),
|
|
stop_loss=(confirmed_row or latest).get('stop_loss'),
|
|
target_price=(confirmed_row or latest).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 []),
|
|
'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 _first_outcome_meta_value(outcomes: list[BacktestOutcome], key: str):
|
|
for outcome in outcomes:
|
|
value = dict(outcome.meta or {}).get(key)
|
|
if value:
|
|
return value
|
|
return None
|
|
|
|
|
|
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']
|
|
time_session_backtest_features = build_time_session_backtest_features(
|
|
{
|
|
'unit': unit,
|
|
'time_session_backtest_features': dict(unit.get('time_session_backtest_features') or {}),
|
|
'session_codes': session_codes,
|
|
'session_filter': 'outside_window' if len(sessions) == 0 else 'covered',
|
|
}
|
|
)
|
|
time_session_kernel_status = evaluate_time_session_kernel_status(time_session_backtest_features)
|
|
return {
|
|
'subject_type': BACKTEST_SUBJECT_PLAN_REPLAY,
|
|
'trading_plan_id': unit.get('trading_plan_id'),
|
|
'setup_playbook_id': unit.get('setup_playbook_id'),
|
|
'doctrine_backtest_cohort': dict(unit.get('doctrine_backtest_cohort') or {}),
|
|
'doctrine_cohort_key': (unit.get('doctrine_backtest_cohort') or {}).get('cohort_key'),
|
|
'doctrine_completeness': (unit.get('doctrine_backtest_cohort') or {}).get('doctrine_completeness'),
|
|
'liquidity_backtest_features': dict(unit.get('liquidity_backtest_features') or {}),
|
|
'liquidity_kernel_status': dict(unit.get('liquidity_kernel_status') or {}),
|
|
'pd_array_backtest_features': dict(unit.get('pd_array_backtest_features') or {}),
|
|
'pd_array_reaction_status': dict(unit.get('pd_array_reaction_status') or {}),
|
|
'time_session_backtest_features': time_session_backtest_features,
|
|
'time_session_kernel_status': time_session_kernel_status,
|
|
'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 resolve_signal_session_codes(*, signal: dict, sessions: list[dict]) -> list[str]:
|
|
session_codes = [row.get('session_code') for row in sessions if row.get('session_code')]
|
|
if session_codes:
|
|
return session_codes
|
|
signal_session = str(signal.get('session_code') or '').strip()
|
|
return [signal_session] if signal_session else []
|
|
|
|
|
|
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,
|
|
entry_price: float,
|
|
entry_triggered: bool,
|
|
cost_config: BacktestCostConfig,
|
|
) -> float:
|
|
if not entry_triggered:
|
|
return 0.0
|
|
if outcome not in {'win', 'loss', 'breakeven'}:
|
|
return 0.0
|
|
|
|
risk_distance = abs(entry_price - signal.stop_loss)
|
|
if risk_distance <= 0:
|
|
return 0.0
|
|
|
|
exit_price = estimate_exit_price(signal=signal, outcome=outcome, entry_price=entry_price)
|
|
rate_cost = (abs(entry_price) + abs(exit_price)) * (cost_config.fee_rate + cost_config.slippage_rate)
|
|
return cost_config.fixed_cost_r + (rate_cost / risk_distance)
|
|
|
|
|
|
def estimate_exit_price(signal: BacktestSignal, outcome: str, entry_price: float) -> float:
|
|
if outcome == 'win' and signal.tp2:
|
|
return signal.tp2
|
|
if outcome == 'loss' and signal.stop_loss:
|
|
return signal.stop_loss
|
|
return entry_price
|
|
|
|
|
|
def compute_outcome_r_multiple(
|
|
outcome: str,
|
|
rr_tp1: float = 0.0,
|
|
rr_tp2: float = 0.0,
|
|
cost_r: float = 0.0,
|
|
tp1_hit: bool = False,
|
|
tp2_hit: bool = False,
|
|
tp1_partial_fraction: float = 0.0,
|
|
stopped_after_tp1: bool = False,
|
|
move_stop_applied: bool = False,
|
|
) -> float:
|
|
"""Convert the current simplified outcome labels into conservative R multiples."""
|
|
if outcome == 'win':
|
|
if tp1_hit and tp1_partial_fraction > 0:
|
|
gross_r = (tp1_partial_fraction * rr_tp1) + ((1.0 - tp1_partial_fraction) * rr_tp2)
|
|
else:
|
|
gross_r = rr_tp2
|
|
elif outcome == 'loss':
|
|
if tp1_hit and tp1_partial_fraction > 0:
|
|
remaining_loss_r = 0.0 if move_stop_applied and stopped_after_tp1 else -(1.0 - tp1_partial_fraction)
|
|
gross_r = (tp1_partial_fraction * rr_tp1) + remaining_loss_r
|
|
else:
|
|
gross_r = -1.0
|
|
elif outcome == 'breakeven':
|
|
if tp1_hit and tp1_partial_fraction > 0:
|
|
gross_r = tp1_partial_fraction * rr_tp1
|
|
else:
|
|
gross_r = 0.0
|
|
else:
|
|
gross_r = 0.0
|
|
|
|
if outcome in {'win', 'loss', 'breakeven'}:
|
|
return gross_r - cost_r
|
|
return gross_r
|
|
|
|
|
|
def has_management_rule(management_rules: list[dict], code: str) -> bool:
|
|
return any(str(item.get("code") or "") == code for item in management_rules)
|
|
|
|
|
|
def resolve_tp1_partial_fraction(management_rules: list[dict]) -> float:
|
|
for item in management_rules:
|
|
if str(item.get("code") or "") != "tp1_partial":
|
|
continue
|
|
try:
|
|
fraction = float(item.get("fraction") or 0.0)
|
|
except (TypeError, ValueError):
|
|
return 0.0
|
|
return min(max(fraction, 0.0), 1.0)
|
|
return 0.0
|
|
|
|
|
|
def classify_backtest_lifecycle_state(
|
|
*,
|
|
outcome: Optional[str],
|
|
tp1_hit: bool,
|
|
tp2_hit: bool,
|
|
invalidated_before_entry: bool,
|
|
) -> Optional[str]:
|
|
if invalidated_before_entry:
|
|
return BACKTEST_LIFECYCLE_INVALIDATED_BEFORE_ENTRY
|
|
if outcome == 'loss':
|
|
return BACKTEST_LIFECYCLE_STOPPED_OUT
|
|
if tp2_hit or outcome == 'win':
|
|
return BACKTEST_LIFECYCLE_TP2_CLOSED
|
|
if tp1_hit or outcome == 'breakeven':
|
|
return BACKTEST_LIFECYCLE_TP1_REACHED
|
|
if outcome == 'missed':
|
|
return BACKTEST_LIFECYCLE_MISSED
|
|
return None
|
|
|
|
|
|
def classify_backtest_lifecycle_state_from_result(backtest_result: Optional[dict]) -> Optional[str]:
|
|
if backtest_result is None:
|
|
return None
|
|
return classify_backtest_lifecycle_state(
|
|
outcome=backtest_result.get('outcome'),
|
|
tp1_hit=bool(backtest_result.get('tp1_hit')),
|
|
tp2_hit=bool(backtest_result.get('tp2_hit')),
|
|
invalidated_before_entry=bool(backtest_result.get('invalidated_before_entry')),
|
|
)
|
|
|
|
|
|
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] = {}
|
|
grouped_by_side: dict[str, dict] = {}
|
|
grouped_by_session: dict[str, dict] = {}
|
|
grouped_by_weekday: dict[str, dict] = {}
|
|
grouped_by_symbol: dict[str, dict] = {}
|
|
grouped_by_timeframe: dict[str, dict] = {}
|
|
grouped_by_doctrine_cohort: dict[str, dict] = {}
|
|
grouped_by_liquidity_target_side: dict[str, dict] = {}
|
|
grouped_by_pd_array_type: dict[str, dict] = {}
|
|
grouped_by_time_session_code: dict[str, dict] = {}
|
|
r_multiples: list[float] = []
|
|
cost_values: list[float] = []
|
|
mfe_r_values: list[float] = []
|
|
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] = {}
|
|
liquidity_kernel_reason_counts: dict[str, int] = {}
|
|
liquidity_kernel_status_counts: dict[str, int] = {}
|
|
pd_array_reaction_reason_counts: dict[str, int] = {}
|
|
pd_array_reaction_status_counts: dict[str, int] = {}
|
|
time_session_kernel_reason_counts: dict[str, int] = {}
|
|
time_session_kernel_status_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
|
|
lifecycle_state = classify_backtest_lifecycle_state(
|
|
outcome=outcome.outcome,
|
|
tp1_hit=outcome.tp1_hit,
|
|
tp2_hit=outcome.tp2_hit,
|
|
invalidated_before_entry=outcome.invalidated_before_entry,
|
|
)
|
|
if lifecycle_state is not None:
|
|
grouped_by_lifecycle_state[lifecycle_state] = grouped_by_lifecycle_state.get(lifecycle_state, 0) + 1
|
|
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']
|
|
weekday = str(outcome.meta.get('weekday') or 'unknown')
|
|
symbol = _extract_symbol_from_outcome(outcome)
|
|
timeframe_bucket = str(outcome.meta.get('signal_timeframe_bucket') or 'unknown')
|
|
doctrine_cohort_key = str(outcome.meta.get('doctrine_cohort_key') or 'missing_playbook::incomplete')
|
|
liquidity_features = dict(outcome.meta.get('liquidity_backtest_features') or {})
|
|
liquidity_target_side = str(liquidity_features.get('liquidity_target_side') or 'missing_liquidity_target')
|
|
liquidity_status = dict(outcome.meta.get('liquidity_kernel_status') or evaluate_liquidity_kernel_status(liquidity_features))
|
|
liquidity_status_key = str(liquidity_status.get('status') or 'unknown')
|
|
liquidity_kernel_status_counts[liquidity_status_key] = liquidity_kernel_status_counts.get(liquidity_status_key, 0) + 1
|
|
for reason in list(liquidity_status.get('reasons') or []):
|
|
liquidity_kernel_reason_counts[str(reason)] = liquidity_kernel_reason_counts.get(str(reason), 0) + 1
|
|
pd_array_features = dict(outcome.meta.get('pd_array_backtest_features') or {})
|
|
pd_array_type = str(pd_array_features.get('pd_array_type') or 'missing_pd_array')
|
|
pd_array_status = dict(outcome.meta.get('pd_array_reaction_status') or evaluate_pd_array_reaction_status(pd_array_features))
|
|
pd_array_status_key = str(pd_array_status.get('status') or 'unknown')
|
|
pd_array_reaction_status_counts[pd_array_status_key] = pd_array_reaction_status_counts.get(pd_array_status_key, 0) + 1
|
|
for reason in list(pd_array_status.get('reasons') or []):
|
|
pd_array_reaction_reason_counts[str(reason)] = pd_array_reaction_reason_counts.get(str(reason), 0) + 1
|
|
time_session_features = dict(outcome.meta.get('time_session_backtest_features') or {})
|
|
time_session_code = str(time_session_features.get('session_code') or (session_codes[0] if session_codes else '') or 'missing_session_window')
|
|
time_session_status = dict(outcome.meta.get('time_session_kernel_status') or evaluate_time_session_kernel_status(time_session_features))
|
|
time_session_status_key = str(time_session_status.get('status') or 'unknown')
|
|
time_session_kernel_status_counts[time_session_status_key] = time_session_kernel_status_counts.get(time_session_status_key, 0) + 1
|
|
for reason in list(time_session_status.get('reasons') or []):
|
|
time_session_kernel_reason_counts[str(reason)] = time_session_kernel_reason_counts.get(str(reason), 0) + 1
|
|
r_multiples.append(float(outcome.meta.get('r_multiple') or 0.0))
|
|
cost_values.append(float(outcome.meta.get('cost_r') or 0.0))
|
|
mfe_r_values.append(float(outcome.meta.get('mfe_r') or 0.0))
|
|
mae_r_values.append(float(outcome.meta.get('mae_r') or 0.0))
|
|
|
|
_accumulate_bucket(grouped_by_side, side, outcome)
|
|
_accumulate_bucket(grouped_by_weekday, weekday, outcome)
|
|
_accumulate_bucket(grouped_by_symbol, symbol, outcome)
|
|
_accumulate_bucket(grouped_by_timeframe, timeframe_bucket, outcome)
|
|
_accumulate_bucket(grouped_by_doctrine_cohort, doctrine_cohort_key, outcome)
|
|
_accumulate_bucket(grouped_by_liquidity_target_side, liquidity_target_side, outcome)
|
|
_accumulate_bucket(grouped_by_pd_array_type, pd_array_type, outcome)
|
|
_accumulate_bucket(grouped_by_time_session_code, time_session_code, outcome)
|
|
for session_code in session_codes:
|
|
_accumulate_bucket(grouped_by_session, str(session_code), outcome)
|
|
|
|
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),
|
|
'loss_rate': _ratio(grouped_by_outcome.get('loss', 0), total),
|
|
'breakeven_rate': _ratio(grouped_by_outcome.get('breakeven', 0), total),
|
|
'miss_rate': _ratio(grouped_by_outcome.get('missed', 0), total),
|
|
'executed_signals': executed_signals,
|
|
'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,
|
|
'liquidity_kernel_status_counts': liquidity_kernel_status_counts,
|
|
'liquidity_kernel_reason_counts': liquidity_kernel_reason_counts,
|
|
'pd_array_reaction_status_counts': pd_array_reaction_status_counts,
|
|
'pd_array_reaction_reason_counts': pd_array_reaction_reason_counts,
|
|
'time_session_kernel_status_counts': time_session_kernel_status_counts,
|
|
'time_session_kernel_reason_counts': time_session_kernel_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),
|
|
'avg_mae_r': _average(mae_r_values),
|
|
'total_r': sum(r_multiples),
|
|
'expectancy_r': _average(r_multiples),
|
|
'profit_factor_r': _profit_factor(r_multiples),
|
|
'max_drawdown_r': compute_max_drawdown_r(r_multiples),
|
|
'max_consecutive_losses': compute_max_consecutive_losses(outcomes),
|
|
'cumulative_r_curve': build_cumulative_r_curve(r_multiples),
|
|
'total_cost_r': sum(cost_values),
|
|
'avg_cost_r': _average(cost_values),
|
|
'by_side': grouped_by_side,
|
|
'by_session': grouped_by_session,
|
|
'by_weekday': grouped_by_weekday,
|
|
'by_symbol': grouped_by_symbol,
|
|
'by_timeframe': grouped_by_timeframe,
|
|
'by_doctrine_cohort': grouped_by_doctrine_cohort,
|
|
'by_liquidity_target_side': grouped_by_liquidity_target_side,
|
|
'by_pd_array_type': grouped_by_pd_array_type,
|
|
'by_time_session_code': grouped_by_time_session_code,
|
|
}
|
|
result['session_performance'] = result['by_session']
|
|
return result
|
|
|
|
|
|
def normalize_timeframe_set(timeframe_set: Optional[dict]) -> dict:
|
|
timeframe_set = dict(timeframe_set or {})
|
|
bias_tf = str(timeframe_set.get('bias_tf') or '1m')
|
|
setup_tf = str(timeframe_set.get('setup_tf') or bias_tf)
|
|
trigger_tf = str(timeframe_set.get('trigger_tf') or setup_tf)
|
|
return {
|
|
'bias_tf': bias_tf,
|
|
'setup_tf': setup_tf,
|
|
'trigger_tf': trigger_tf,
|
|
}
|
|
|
|
|
|
def extract_target_plan_from_signal_row(signal: dict) -> dict:
|
|
return dict(signal.get('target_plan') or {})
|
|
|
|
|
|
def extract_management_rules_from_signal_row(signal: dict) -> list[dict]:
|
|
target_plan = dict(signal.get('target_plan') or {})
|
|
return list(target_plan.get('management_rules') or [])
|
|
|
|
|
|
def build_timeframe_bucket(*, bias_tf: str, setup_tf: str, trigger_tf: str) -> str:
|
|
return f"{bias_tf}/{setup_tf}/{trigger_tf}"
|
|
|
|
|
|
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] = []
|
|
|
|
total_signals = int(summary.get('total_signals') or 0)
|
|
expectancy_r = float(summary.get('expectancy_r') or 0.0)
|
|
profit_factor_r = summary.get('profit_factor_r')
|
|
max_drawdown_r = float(summary.get('max_drawdown_r') or 0.0)
|
|
max_consecutive_losses = int(summary.get('max_consecutive_losses') or 0)
|
|
executed_signals = int(summary.get('executed_signals') or 0)
|
|
|
|
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')
|
|
if expectancy_r <= gate.min_expectancy_r:
|
|
reasons.append('expectancy_below_threshold')
|
|
if profit_factor_r is None or float(profit_factor_r) < gate.min_profit_factor_r:
|
|
reasons.append('profit_factor_below_threshold')
|
|
if max_drawdown_r > gate.max_drawdown_r:
|
|
reasons.append('drawdown_above_threshold')
|
|
if max_consecutive_losses > gate.max_consecutive_losses:
|
|
reasons.append('consecutive_losses_above_threshold')
|
|
|
|
return {
|
|
'passed': len(reasons) == 0,
|
|
'reasons': reasons,
|
|
'gate': gate.__dict__,
|
|
'sample_basis': sample_basis,
|
|
'metrics': {
|
|
'total_signals': total_signals,
|
|
'executed_signals': executed_signals,
|
|
'expectancy_r': expectancy_r,
|
|
'profit_factor_r': profit_factor_r,
|
|
'max_drawdown_r': max_drawdown_r,
|
|
'max_consecutive_losses': max_consecutive_losses,
|
|
},
|
|
}
|
|
|
|
|
|
def _accumulate_bucket(container: dict, key: str, outcome: BacktestOutcome) -> None:
|
|
bucket = container.setdefault(
|
|
key,
|
|
{
|
|
'count': 0,
|
|
'wins': 0,
|
|
'losses': 0,
|
|
'breakevens': 0,
|
|
'misses': 0,
|
|
'executed': 0,
|
|
'executed_total_r': 0.0,
|
|
'avg_mfe_sum': 0.0,
|
|
'avg_mae_sum': 0.0,
|
|
'avg_mfe_r_sum': 0.0,
|
|
'avg_mae_r_sum': 0.0,
|
|
'current_loss_streak': 0,
|
|
'max_consecutive_losses': 0,
|
|
},
|
|
)
|
|
bucket['count'] += 1
|
|
if is_executed_outcome(outcome):
|
|
bucket['executed'] += 1
|
|
if outcome.outcome == 'win':
|
|
bucket['wins'] += 1
|
|
bucket['current_loss_streak'] = 0
|
|
elif outcome.outcome == 'loss':
|
|
bucket['losses'] += 1
|
|
bucket['current_loss_streak'] += 1
|
|
elif outcome.outcome == 'breakeven':
|
|
bucket['breakevens'] += 1
|
|
bucket['current_loss_streak'] = 0
|
|
else:
|
|
bucket['misses'] += 1
|
|
bucket['current_loss_streak'] = 0
|
|
bucket['max_consecutive_losses'] = max(bucket['max_consecutive_losses'], bucket['current_loss_streak'])
|
|
bucket['avg_mfe_sum'] += outcome.mfe
|
|
bucket['avg_mae_sum'] += outcome.mae
|
|
bucket['avg_mfe_r_sum'] += float(outcome.meta.get('mfe_r') or 0.0)
|
|
bucket['avg_mae_r_sum'] += float(outcome.meta.get('mae_r') or 0.0)
|
|
r_multiple = float(outcome.meta.get('r_multiple') or 0.0)
|
|
bucket['total_r'] = bucket.get('total_r', 0.0) + r_multiple
|
|
if is_executed_outcome(outcome):
|
|
bucket['executed_total_r'] += r_multiple
|
|
cost_r = float(outcome.meta.get('cost_r') or 0.0)
|
|
bucket['total_cost_r'] = bucket.get('total_cost_r', 0.0) + cost_r
|
|
bucket['win_rate'] = _ratio(bucket['wins'], bucket['count'])
|
|
bucket['executed_rate'] = _ratio(bucket['executed'], bucket['count'])
|
|
bucket['executed_expectancy_r'] = bucket['executed_total_r'] / bucket['executed'] if bucket['executed'] else 0.0
|
|
bucket['avg_mfe'] = bucket['avg_mfe_sum'] / bucket['count'] if bucket['count'] else 0.0
|
|
bucket['avg_mae'] = bucket['avg_mae_sum'] / bucket['count'] if bucket['count'] else 0.0
|
|
bucket['avg_mfe_r'] = bucket['avg_mfe_r_sum'] / bucket['count'] if bucket['count'] else 0.0
|
|
bucket['avg_mae_r'] = bucket['avg_mae_r_sum'] / bucket['count'] if bucket['count'] else 0.0
|
|
bucket['expectancy_r'] = bucket['total_r'] / bucket['count'] if bucket['count'] else 0.0
|
|
bucket['avg_cost_r'] = bucket['total_cost_r'] / bucket['count'] if bucket['count'] else 0.0
|
|
curve = bucket.setdefault('cumulative_r_curve', [])
|
|
curve.append(bucket['total_r'])
|
|
bucket['max_drawdown_r'] = compute_drawdown_from_curve(curve)
|
|
|
|
|
|
def is_executed_outcome(outcome: BacktestOutcome) -> bool:
|
|
return outcome.outcome in {'win', 'loss', 'breakeven'} or bool(outcome.meta.get('entry_triggered'))
|
|
|
|
|
|
def _extract_symbol_from_outcome(outcome: BacktestOutcome) -> str:
|
|
symbol = outcome.meta.get('signal_symbol')
|
|
if symbol:
|
|
return str(symbol)
|
|
instrument_id = outcome.meta.get('instrument_id')
|
|
if instrument_id is not None:
|
|
return f"instrument:{instrument_id}"
|
|
return 'UNKNOWN'
|
|
|
|
|
|
def _ratio(numerator: int, denominator: int) -> float:
|
|
return numerator / denominator if denominator else 0.0
|
|
|
|
|
|
def _average(values: list[float]) -> float:
|
|
return sum(values) / len(values) if values else 0.0
|
|
|
|
|
|
def build_cumulative_r_curve(values: list[float]) -> list[float]:
|
|
curve: list[float] = []
|
|
running = 0.0
|
|
for value in values:
|
|
running += value
|
|
curve.append(running)
|
|
return curve
|
|
|
|
|
|
def compute_max_drawdown_r(values: list[float]) -> float:
|
|
return compute_drawdown_from_curve(build_cumulative_r_curve(values))
|
|
|
|
|
|
def compute_max_consecutive_losses(outcomes: list[BacktestOutcome]) -> int:
|
|
current = 0
|
|
longest = 0
|
|
for outcome in outcomes:
|
|
if outcome.outcome == 'loss':
|
|
current += 1
|
|
longest = max(longest, current)
|
|
else:
|
|
current = 0
|
|
return longest
|
|
|
|
|
|
def compute_drawdown_from_curve(curve: list[float]) -> float:
|
|
peak = 0.0
|
|
max_drawdown = 0.0
|
|
for value in curve:
|
|
peak = max(peak, value)
|
|
max_drawdown = max(max_drawdown, peak - value)
|
|
return max_drawdown
|
|
|
|
|
|
def _profit_factor(values: list[float]) -> Optional[float]:
|
|
gross_profit = sum(value for value in values if value > 0)
|
|
gross_loss = abs(sum(value for value in values if value < 0))
|
|
if gross_loss == 0:
|
|
return None if gross_profit == 0 else gross_profit
|
|
return gross_profit / gross_loss
|
|
|
|
|
|
def _extract_weekday(ts) -> str:
|
|
if isinstance(ts, datetime):
|
|
return ts.strftime('%A')
|
|
if isinstance(ts, str):
|
|
normalized = ts.replace('Z', '+00:00')
|
|
return datetime.fromisoformat(normalized).strftime('%A')
|
|
return 'unknown'
|