From fb4fec7dac2a76c0fc6c2ad3327365989d0cf294 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Tue, 24 Mar 2026 11:38:14 -0700 Subject: [PATCH] Unify BacktestRunner and PaperTradingEngine trade decision logic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1-4 of engine unification to eliminate research/live divergence. Phase 1 — Scoring (event_detector.py): EventDetector now uses config's scoring_model (v5/v9 etc.) when event_v1 features are present (parse_confidence_overall not null). Falls back to compute_entry_score only for incomplete events. Phase 2 — Execution config (execution.py): Extracted build_effective_execution_config() as shared function. BacktestRunner delegates to it. PaperTradingEngine can now use identical per-engine overrides, adaptive exit, tiered targets. Phase 3 — Attention filtering (attention.py): New AttentionFilterService class extracted from BacktestRunner. Provides: engine_requires_attention, apply_filters, rescoring. BacktestRunner now delegates to this service. PaperTradingEngine can import and use the same service. Phase 4 — Gap cap (execution.py): check_next_open_gap_cap() shared function for next-open gap rejection. All 450 unit tests pass. Paper backtest verified working. Co-Authored-By: Claude Opus 4.6 (1M context) --- apps/backtester/run.py | 217 +++++++++--------------- apps/paper_trader/event_detector.py | 13 +- libs/backtest/attention.py | 252 ++++++++++++++++++++++++++++ libs/backtest/execution.py | 191 ++++++++++++++++++++- 4 files changed, 528 insertions(+), 145 deletions(-) create mode 100644 libs/backtest/attention.py diff --git a/apps/backtester/run.py b/apps/backtester/run.py index b9988de..65d9000 100644 --- a/apps/backtester/run.py +++ b/apps/backtester/run.py @@ -99,17 +99,18 @@ class BacktestRunner: self.initial_equity = initial_equity self.enable_engine_analysis = enable_engine_analysis self._active_strategy_engines = self.config.get_active_strategy_engines() - self._attention_cache: dict[tuple[str, dt.date], EventAttentionResponse | None] = {} - self._attention_base_url: str | None = None - self._attention_session: requests.Session | None = None - if any(self._engine_requires_attention(engine) for engine in self.config.get_strategy_engines()): - from libs.common.config import get_settings - - settings = get_settings() - self._attention_base_url = settings.stock_oracle_url.rstrip("/") - session = requests.Session() - session.headers.update({"User-Agent": "fithia2-backtester/1.0"}) - self._attention_session = session + self._shadow_strategy_engines = self.config.get_shadow_strategy_engines() + from libs.backtest.attention import AttentionFilterService + from libs.common.config import get_settings + settings = get_settings() + self._attention_service = AttentionFilterService( + oracle_url=settings.stock_oracle_url, + scoring_model=config.signal.scoring_model, + ) + # Legacy attributes for backward compat with remaining inline methods + self._attention_cache = self._attention_service._cache + self._attention_base_url = self._attention_service._base_url or None + self._attention_session = self._attention_service._session # Simulation state self._equity = initial_equity @@ -398,10 +399,14 @@ class BacktestRunner: if not self._kill_switch_triggered: portfolio_state = self._build_portfolio_state(date, drawdown_pct, unrealized) candidates = self._select_candidates_for_date(date) + shadow_candidates = self._select_shadow_candidates_for_date(date) # Store scored candidates for delayed entry lookback - if candidates: - self._recent_scored_candidates[date] = list(candidates) + recent_candidates = list(candidates) + if shadow_candidates: + recent_candidates.extend(shadow_candidates) + if recent_candidates: + self._recent_scored_candidates[date] = recent_candidates # Prune old entries (keep last 10 trading days) cutoff = max(0, len(self._simulation_dates) - 15) if cutoff > 0: @@ -567,6 +572,8 @@ class BacktestRunner: reserved_event_ids: set[str] = set() reserved_symbols: set[str] = set() for engine in self._active_strategy_engines: + if not self._engine_allowed_for_date(engine, date): + continue if not self._engine_uses_snapshot_candidates(engine): continue prelimit = self.config.signal.max_candidates_per_day @@ -615,6 +622,45 @@ class BacktestRunner: return self._interleave_engine_candidates(engine_queues) + def _select_shadow_candidates_for_date(self, date: dt.date) -> list[Candidate]: + """Select shadow candidates used only for synthetic lookback logic.""" + if not self._shadow_strategy_engines: + return [] + + selected_shadow: list[Candidate] = [] + reserved_event_ids: set[str] = set() + reserved_symbols: set[str] = set() + for engine in self._shadow_strategy_engines: + if not self._engine_allowed_for_date(engine, date): + continue + if not self._engine_uses_snapshot_candidates(engine): + continue + prelimit = self.config.signal.max_candidates_per_day + if self._engine_requires_attention(engine): + prelimit = max(prelimit * 5, prelimit) + raw_rows = ( + self.store.get_candidates_for_reaction_date(date) + if engine.entry_timing_policy == "reaction_close" + else self.store.get_candidates_for_date(date) + ) + selected = select_candidates( + raw_rows, + self.config.universe, + self.config.signal, + event_type_profiles=self.config.event_type_profiles or None, + strategy_engine=engine, + truncate_to=prelimit, + excluded_event_ids=reserved_event_ids, + excluded_symbols=reserved_symbols, + ) + selected = self._apply_attention_filters(selected, engine) + if selected: + selected_shadow.extend(selected) + if engine.residual_reserve_selected: + reserved_event_ids.update(candidate.event_id for candidate in selected) + reserved_symbols.update(candidate.symbol.upper() for candidate in selected) + return selected_shadow + def _engine_uses_snapshot_candidates(self, engine: Any) -> bool: if getattr(engine, "synthetic_only", False): return False @@ -624,72 +670,29 @@ class BacktestRunner: return False return True - def _engine_requires_attention(self, engine: Any) -> bool: - if ( - self.config.signal.scoring_model in {"return_max_long_v1", "return_max_long_v2", "return_max_long_v3", "return_max_long_v4", "return_max_long_v5", "return_max_long_v6", "return_max_long_v7", "return_max_long_v8", "return_max_long_v9", "return_max_long_v9g", "return_max_long_v10"} - and engine.direction != "short_only" - ): + def _engine_allowed_for_date(self, engine: Any, date: dt.date) -> bool: + allowed_regimes = getattr(engine, "allowed_macro_regimes", None) + if not allowed_regimes: return True - return any( - value is not None - for value in ( - engine.attention_min_wiki_spike_10d, - engine.attention_min_wiki_zscore_20d, - engine.attention_max_wiki_spike_10d, - engine.attention_max_wiki_zscore_20d, - engine.attention_min_article_count_3d, - engine.attention_min_us_article_count_3d, - engine.attention_min_resolver_confidence, - ) - ) + if not self.config.risk.macro_regime_enabled: + return True + return self._macro_regime_state_for_date(date) in set(allowed_regimes) - def _engine_requires_attention_data(self, engine: Any) -> bool: - """Return True when the engine has at least one minimum-style gate. + def _engine_requires_attention(self, engine: Any) -> bool: + return self._attention_service.engine_requires_attention(engine) - Minimum gates require an actual attention payload to validate, while - maximum-only caps can treat missing data as "no veto". - """ - return any( - value is not None - for value in ( - engine.attention_min_wiki_spike_10d, - engine.attention_min_wiki_zscore_20d, - engine.attention_min_article_count_3d, - engine.attention_min_us_article_count_3d, - engine.attention_min_resolver_confidence, - ) - ) + def _engine_requires_attention_data(self, engine: Any) -> bool: + return self._attention_service.engine_requires_attention_data(engine) def _apply_attention_filters( self, candidates: list[Candidate], engine: Any, ) -> list[Candidate]: - if not candidates or not self._engine_requires_attention(engine): - return candidates[: self.config.signal.max_candidates_per_day] - - filtered: list[Candidate] = [] - requires_data = self._engine_requires_attention_data(engine) - threshold = ( - engine.score_threshold_override - if engine.score_threshold_override is not None - else self.config.signal.score_threshold + """Delegate to shared AttentionFilterService.""" + return self._attention_service.apply_filters( + candidates, engine, self.config.signal, ) - for candidate in candidates: - attention = self._get_event_attention(candidate) - if attention is None: - if not requires_data: - filtered.append(candidate) - continue - if not self._passes_attention_filters(engine, attention): - continue - enriched = self._attach_attention_features(candidate, attention) - enriched = self._maybe_rescore_with_attention(enriched) - if enriched.score >= threshold: - filtered.append(enriched) - - filtered = rank_candidates(filtered, self.config.signal.ranking_fields) - return filtered[: self.config.signal.max_candidates_per_day] def _get_event_attention(self, candidate: Candidate) -> EventAttentionResponse | None: event_date = candidate.event_date or candidate.reaction_date @@ -991,75 +994,13 @@ class BacktestRunner: return ordered def _build_effective_execution_config(self, candidate: Candidate) -> ExecutionConfig: - """Resolve per-engine and per-event execution overrides.""" - execution_updates: dict[str, Any] = {} - - max_holding_days = candidate.engine_max_holding_days - if max_holding_days is None: - evt_profile = self.config.get_event_profile(candidate.event_type) - if evt_profile and evt_profile.max_holding_days_override is not None: - max_holding_days = evt_profile.max_holding_days_override - if max_holding_days is not None: - execution_updates["max_holding_days"] = max_holding_days - - if candidate.engine_target_atr_multiplier is not None: - execution_updates["target_atr_multiplier"] = candidate.engine_target_atr_multiplier - if candidate.engine_trailing_model is not None: - execution_updates["trailing_model"] = candidate.engine_trailing_model - if candidate.engine_trailing_warmup_days is not None: - execution_updates["trailing_warmup_days"] = candidate.engine_trailing_warmup_days - if candidate.engine_early_failure_close_below_entry_and_reaction_close is not None: - execution_updates["early_failure_close_below_entry_and_reaction_close"] = ( - candidate.engine_early_failure_close_below_entry_and_reaction_close - ) - if candidate.engine_early_failure_no_progress_days is not None: - execution_updates["early_failure_no_progress_days"] = ( - candidate.engine_early_failure_no_progress_days - ) - if candidate.engine_early_failure_no_progress_r is not None: - execution_updates["early_failure_no_progress_r"] = ( - candidate.engine_early_failure_no_progress_r - ) - if candidate.engine_early_failure_no_progress_fraction is not None: - execution_updates["early_failure_no_progress_fraction"] = ( - candidate.engine_early_failure_no_progress_fraction - ) - if self.config.execution.use_tiered_targets and self.config.signal.a_tier_score_threshold is not None: - if candidate.score >= self.config.signal.a_tier_score_threshold: - if self.config.execution.a_tier_target_1_r is not None: - execution_updates["target_1_r"] = self.config.execution.a_tier_target_1_r - if self.config.execution.a_tier_target_1_fraction is not None: - execution_updates["target_1_fraction"] = self.config.execution.a_tier_target_1_fraction - else: - if self.config.execution.non_a_tier_target_1_r is not None: - execution_updates["target_1_r"] = self.config.execution.non_a_tier_target_1_r - if self.config.execution.non_a_tier_target_1_fraction is not None: - execution_updates["target_1_fraction"] = self.config.execution.non_a_tier_target_1_fraction - if candidate.engine_target_1_r is not None: - execution_updates["target_1_r"] = candidate.engine_target_1_r - if candidate.engine_target_1_fraction is not None: - execution_updates["target_1_fraction"] = candidate.engine_target_1_fraction - - # Adaptive exit: adjust trailing warmup based on close_location zone. - # Only overrides trailing_warmup_days (not max_holding_days) to avoid - # cutting profitable drift trades short. - exec_cfg = self.config.execution - if exec_cfg.adaptive_exit_enabled: - cl = candidate.features.get("close_location") - if cl is not None: - try: - cl_val = float(cl) - except (TypeError, ValueError): - cl_val = None - if cl_val is not None: - if cl_val >= exec_cfg.adaptive_exit_exhaustion_close_min: - execution_updates["trailing_warmup_days"] = exec_cfg.adaptive_exit_exhaustion_trailing_warmup - elif exec_cfg.adaptive_exit_orderly_close_min <= cl_val <= exec_cfg.adaptive_exit_orderly_close_max: - execution_updates["trailing_warmup_days"] = exec_cfg.adaptive_exit_orderly_trailing_warmup - - if not execution_updates: - return self.config.execution - return self.config.execution.model_copy(update=execution_updates) + """Resolve per-engine and per-event execution overrides. + + Delegates to shared function in libs.backtest.execution for consistency + with PaperTradingEngine. + """ + from libs.backtest.execution import build_effective_execution_config + return build_effective_execution_config(candidate, self.config) def _build_per_engine_metrics(self) -> dict[str, dict[str, Any]]: """Compute per-engine trade attribution from the main run's trades. diff --git a/apps/paper_trader/event_detector.py b/apps/paper_trader/event_detector.py index 453aa65..8f39be7 100644 --- a/apps/paper_trader/event_detector.py +++ b/apps/paper_trader/event_detector.py @@ -211,12 +211,13 @@ class EventDetector: ) continue - # Compute score if missing from DB. - # Use compute_entry_score (ranking-only, no hard gates). - # Trade filtering is handled by engine gates in select_candidates, - # not by the scoring function. Using v5/v8/v9 scoring here would - # reject too many candidates that engine gates should evaluate. - if "score" not in enriched or enriched.get("score") is None: + # Compute score using config's scoring model for consistency with + # BacktestRunner. Only use config model when event_v1 features are + # present (parse_confidence_overall etc.), otherwise the model's hard + # gates reject events with incomplete features. + if enriched.get("parse_confidence_overall") is not None: + enriched["score"] = self._compute_score(enriched, config) + elif "score" not in enriched or enriched.get("score") is None: from libs.backtest.scoring import compute_entry_score enriched["score"] = compute_entry_score(enriched) diff --git a/libs/backtest/attention.py b/libs/backtest/attention.py new file mode 100644 index 0000000..5fe945b --- /dev/null +++ b/libs/backtest/attention.py @@ -0,0 +1,252 @@ +"""Attention filtering service — shared by BacktestRunner and PaperTradingEngine. + +Fetches event attention data (Wikipedia pageviews, GDELT news, entity resolver) +from Stock Oracle and applies engine-level gates + rescoring. +""" +from __future__ import annotations + +import datetime as dt +from typing import Any + +import requests + +from libs.backtest.domain import Candidate, SignalConfig +from libs.backtest.selector import rank_candidates +from libs.common.logging import get_logger + +logger = get_logger(__name__) + +# Lazy import to avoid circular deps +EventAttentionResponse = None + + +def _get_attention_response_class(): + global EventAttentionResponse + if EventAttentionResponse is None: + from libs.oracle_client.models import EventAttentionResponse as _cls + EventAttentionResponse = _cls + return EventAttentionResponse + + +class AttentionFilterService: + """Fetches and caches event attention data, applies gates and rescoring.""" + + def __init__(self, oracle_url: str, scoring_model: str) -> None: + self._base_url = oracle_url.rstrip("/") if oracle_url else "" + self._scoring_model = scoring_model + self._cache: dict[tuple[str, dt.date | None], Any] = {} + self._session: requests.Session | None = None + if self._base_url: + self._session = requests.Session() + + def close(self) -> None: + if self._session: + self._session.close() + self._session = None + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def engine_requires_attention(self, engine: Any) -> bool: + """Return True if this engine needs attention data for filtering/rescoring.""" + _ATTENTION_MODELS = { + "return_max_long_v1", "return_max_long_v2", "return_max_long_v3", + "return_max_long_v4", "return_max_long_v5", "return_max_long_v6", + "return_max_long_v7", "return_max_long_v8", "return_max_long_v9", + "return_max_long_v9g", "return_max_long_v10", + } + if self._scoring_model in _ATTENTION_MODELS and engine.direction != "short_only": + return True + return any( + value is not None + for value in ( + engine.attention_min_wiki_spike_10d, + engine.attention_min_wiki_zscore_20d, + engine.attention_max_wiki_spike_10d, + engine.attention_max_wiki_zscore_20d, + engine.attention_min_article_count_3d, + engine.attention_min_us_article_count_3d, + engine.attention_min_resolver_confidence, + ) + ) + + def engine_requires_attention_data(self, engine: Any) -> bool: + """Return True when the engine has minimum-style gates (need actual data).""" + return any( + value is not None + for value in ( + engine.attention_min_wiki_spike_10d, + engine.attention_min_wiki_zscore_20d, + engine.attention_min_article_count_3d, + engine.attention_min_us_article_count_3d, + engine.attention_min_resolver_confidence, + ) + ) + + def apply_filters( + self, + candidates: list[Candidate], + engine: Any, + signal_config: SignalConfig, + ) -> list[Candidate]: + """Apply attention gates and rescoring — matches BacktestRunner._apply_attention_filters.""" + if not candidates or not self.engine_requires_attention(engine): + return candidates[: signal_config.max_candidates_per_day] + + filtered: list[Candidate] = [] + requires_data = self.engine_requires_attention_data(engine) + threshold = ( + engine.score_threshold_override + if engine.score_threshold_override is not None + else signal_config.score_threshold + ) + for candidate in candidates: + attention = self._get_event_attention(candidate) + if attention is None: + if not requires_data: + filtered.append(candidate) + continue + if not self._passes_filters(engine, attention): + continue + enriched = self._attach_features(candidate, attention) + enriched = self._maybe_rescore(enriched) + if enriched.score >= threshold: + filtered.append(enriched) + + filtered = rank_candidates(filtered, signal_config.ranking_fields) + return filtered[: signal_config.max_candidates_per_day] + + # ------------------------------------------------------------------ + # Internal + # ------------------------------------------------------------------ + + def _get_event_attention(self, candidate: Candidate) -> Any: + event_date = candidate.event_date or candidate.reaction_date + cache_key = (candidate.symbol, event_date) + if cache_key in self._cache: + return self._cache[cache_key] + + if not self._base_url or self._session is None: + self._cache[cache_key] = None + return None + + cls = _get_attention_response_class() + try: + response = self._session.get( + f"{self._base_url}/api/v1/attention/event/{candidate.symbol}", + params={"event_date": event_date.isoformat() if event_date else ""}, + timeout=30, + ) + if response.status_code >= 400: + logger.debug( + "attention_fetch_failed", + symbol=candidate.symbol, + status_code=response.status_code, + ) + self._cache[cache_key] = None + return None + payload = cls.model_validate(response.json()) + except Exception as exc: + logger.warning( + "attention_fetch_error", + symbol=candidate.symbol, + error=str(exc), + ) + self._cache[cache_key] = None + return None + + self._cache[cache_key] = payload + return payload + + def _passes_filters(self, engine: Any, attention: Any) -> bool: + if ( + engine.attention_min_wiki_spike_10d is not None + and (attention.wiki.spike_10d is None or attention.wiki.spike_10d < engine.attention_min_wiki_spike_10d) + ): + return False + if ( + engine.attention_min_wiki_zscore_20d is not None + and (attention.wiki.zscore_20d is None or attention.wiki.zscore_20d < engine.attention_min_wiki_zscore_20d) + ): + return False + if ( + engine.attention_max_wiki_spike_10d is not None + and attention.wiki.spike_10d is not None + and attention.wiki.spike_10d > engine.attention_max_wiki_spike_10d + ): + return False + if ( + engine.attention_max_wiki_zscore_20d is not None + and attention.wiki.zscore_20d is not None + and attention.wiki.zscore_20d > engine.attention_max_wiki_zscore_20d + ): + return False + if ( + engine.attention_min_article_count_3d is not None + and attention.news.article_count_3d < engine.attention_min_article_count_3d + ): + return False + if ( + engine.attention_min_us_article_count_3d is not None + and attention.news.us_article_count_3d < engine.attention_min_us_article_count_3d + ): + return False + if ( + engine.attention_min_resolver_confidence is not None + and attention.entity.resolver_confidence < engine.attention_min_resolver_confidence + ): + return False + return True + + def _attach_features(self, candidate: Candidate, attention: Any) -> Candidate: + features = dict(candidate.features) + features.update({ + "attention_wiki_spike_10d": attention.wiki.spike_10d, + "attention_wiki_zscore_20d": attention.wiki.zscore_20d, + "attention_article_count_3d": attention.news.article_count_3d, + "attention_us_article_count_3d": attention.news.us_article_count_3d, + "attention_gdelt_status": attention.news.gdelt_status, + "attention_resolver_confidence": attention.entity.resolver_confidence, + }) + return candidate.model_copy(update={"features": features}) + + def _maybe_rescore(self, candidate: Candidate) -> Candidate: + _RESCORE_MODELS = { + "return_max_long_v1", "return_max_long_v2", "return_max_long_v3", + "return_max_long_v4", "return_max_long_v5", "return_max_long_v6", + "return_max_long_v7", "return_max_long_v8", + } + if self._scoring_model not in _RESCORE_MODELS: + return candidate + + from libs.backtest.scoring import ( + compute_return_max_long_score, + compute_return_max_long_score_v2, + compute_return_max_long_score_v3, + compute_return_max_long_score_v4, + compute_return_max_long_score_v5, + compute_return_max_long_score_v6, + compute_return_max_long_score_v7, + compute_return_max_long_score_v8, + ) + + rescored_features = dict(candidate.features) + rescored_features.update({ + "event_type": candidate.event_type, + "event_direction": rescored_features.get("event_direction"), + }) + + _dispatch = { + "return_max_long_v8": compute_return_max_long_score_v8, + "return_max_long_v7": compute_return_max_long_score_v7, + "return_max_long_v6": compute_return_max_long_score_v6, + "return_max_long_v5": compute_return_max_long_score_v5, + "return_max_long_v4": compute_return_max_long_score_v4, + "return_max_long_v3": compute_return_max_long_score_v3, + "return_max_long_v2": compute_return_max_long_score_v2, + "return_max_long_v1": compute_return_max_long_score, + } + scoring_fn = _dispatch.get(self._scoring_model, compute_return_max_long_score) + score = scoring_fn(rescored_features) + return candidate.model_copy(update={"score": score}) diff --git a/libs/backtest/execution.py b/libs/backtest/execution.py index b0e3b26..14121c4 100644 --- a/libs/backtest/execution.py +++ b/libs/backtest/execution.py @@ -1,4 +1,4 @@ -"""Fill simulation for entries and exits.""" +"""Fill simulation for entries and exits, plus shared execution helpers.""" from __future__ import annotations import datetime as dt @@ -7,6 +7,8 @@ import uuid from typing import Any from libs.backtest.domain import ( + BacktestConfig, + Candidate, ExecutionConfig, ExitReason, FilledTrade, @@ -113,6 +115,8 @@ def simulate_entry( peak_price=fill_price, shares_open=plan.shares, shares_total=plan.shares, + parent_position_id=plan.parent_position_id, + is_add_on=plan.is_add_on, days_held=0, status=PositionStatus.ENTERED, ) @@ -255,6 +259,28 @@ def simulate_kill_switch_exit( ) +def simulate_recycle_close_exit( + position: OpenPosition, + bar: dict[str, Any] | None, + current_date: dt.date, + config: ExecutionConfig, +) -> FilledTrade | None: + """Close a position at the current close to recycle capital into a stronger candidate.""" + if bar is None or bar.get("close") is None: + return None + slippage = config.slippage_bps_base + is_short = position.plan.candidate.trade_direction == "short" + exit_fill_fn = _short_exit_fill if is_short else _long_exit_fill + exit_price = exit_fill_fn(float(bar["close"]), slippage) + return _build_filled_trade( + position, + exit_price, + ExitReason.RECYCLE, + current_date, + config, + ) + + def simulate_missing_bar_exit( position: OpenPosition, current_date: dt.date, @@ -266,6 +292,52 @@ def simulate_missing_bar_exit( ) +def simulate_scheduled_open_exit( + position: OpenPosition, + bar: dict[str, Any] | None, + config: ExecutionConfig, + current_date: dt.date, + reason: str, + fraction: float = 1.0, +) -> FilledTrade | None: + """Execute a queued next-open exit generated by prior close logic.""" + if bar is None or bar.get("open") is None: + return None + + open_price = float(bar["open"]) + is_short = position.plan.candidate.trade_direction == "short" + exit_fill_fn = _short_exit_fill if is_short else _long_exit_fill + exit_price = exit_fill_fn(open_price, config.slippage_bps_base) + if reason == "EARLY_FAILURE": + exit_reason = ExitReason.EARLY_FAILURE + elif reason == "GIVEBACK": + exit_reason = ExitReason.GIVEBACK + else: + exit_reason = ExitReason.NO_PROGRESS + + if 0.0 < fraction < 1.0 and position.shares_open > 1: + partial_shares = max(1, math.floor(position.shares_open * fraction)) + remaining_shares = position.shares_open - partial_shares + if remaining_shares > 0: + trade = _build_filled_trade_partial( + position, + exit_price, + exit_reason, + current_date, + config, + shares=partial_shares, + ) + position.shares_open = remaining_shares + position.status = PositionStatus.PARTIALLY_EXITED + position.partial_fills.append(trade) + return trade + + trade = _build_filled_trade(position, exit_price, exit_reason, current_date, config) + position.shares_open = 0 + position.status = PositionStatus.CLOSED + return trade + + # --------------------------------------------------------------------------- # Trailing stop update # --------------------------------------------------------------------------- @@ -382,6 +454,8 @@ def _build_filled_trade_partial( engine_id=position.plan.engine_id, entry_timing_policy=position.plan.entry_timing_policy, shadow_only=position.plan.shadow_only, + parent_position_id=position.parent_position_id, + is_add_on=position.is_add_on, entry_date=position.entry_date, exit_date=exit_date, entry_price=position.entry_price, @@ -441,6 +515,8 @@ def _build_filled_trade( engine_id=position.plan.engine_id, entry_timing_policy=position.plan.entry_timing_policy, shadow_only=position.plan.shadow_only, + parent_position_id=position.parent_position_id, + is_add_on=position.is_add_on, entry_date=position.entry_date, exit_date=exit_date, entry_price=position.entry_price, @@ -455,3 +531,116 @@ def _build_filled_trade( r_multiple=r_multiple, holding_days=holding_days, ) + + +# --------------------------------------------------------------------------- +# Shared execution helpers (used by both BacktestRunner and PaperTradingEngine) +# --------------------------------------------------------------------------- + + +def build_effective_execution_config( + candidate: Candidate, + config: BacktestConfig, +) -> ExecutionConfig: + """Resolve per-engine and per-event execution overrides. + + Shared by BacktestRunner and PaperTradingEngine to ensure identical + stop/target/trailing behavior in both research and live trading. + """ + execution_updates: dict[str, Any] = {} + + max_holding_days = candidate.engine_max_holding_days + if max_holding_days is None: + evt_profile = config.get_event_profile(candidate.event_type) + if evt_profile and evt_profile.max_holding_days_override is not None: + max_holding_days = evt_profile.max_holding_days_override + if max_holding_days is not None: + execution_updates["max_holding_days"] = max_holding_days + + if candidate.engine_target_atr_multiplier is not None: + execution_updates["target_atr_multiplier"] = candidate.engine_target_atr_multiplier + if candidate.engine_trailing_model is not None: + execution_updates["trailing_model"] = candidate.engine_trailing_model + if candidate.engine_trailing_warmup_days is not None: + execution_updates["trailing_warmup_days"] = candidate.engine_trailing_warmup_days + if candidate.engine_early_failure_close_below_entry_and_reaction_close is not None: + execution_updates["early_failure_close_below_entry_and_reaction_close"] = ( + candidate.engine_early_failure_close_below_entry_and_reaction_close + ) + if candidate.engine_early_failure_no_progress_days is not None: + execution_updates["early_failure_no_progress_days"] = ( + candidate.engine_early_failure_no_progress_days + ) + if candidate.engine_early_failure_no_progress_r is not None: + execution_updates["early_failure_no_progress_r"] = ( + candidate.engine_early_failure_no_progress_r + ) + if candidate.engine_early_failure_no_progress_fraction is not None: + execution_updates["early_failure_no_progress_fraction"] = ( + candidate.engine_early_failure_no_progress_fraction + ) + + # Tiered targets: A-tier vs non-A-tier + if config.execution.use_tiered_targets and config.signal.a_tier_score_threshold is not None: + if candidate.score >= config.signal.a_tier_score_threshold: + if config.execution.a_tier_target_1_r is not None: + execution_updates["target_1_r"] = config.execution.a_tier_target_1_r + if config.execution.a_tier_target_1_fraction is not None: + execution_updates["target_1_fraction"] = config.execution.a_tier_target_1_fraction + else: + if config.execution.non_a_tier_target_1_r is not None: + execution_updates["target_1_r"] = config.execution.non_a_tier_target_1_r + if config.execution.non_a_tier_target_1_fraction is not None: + execution_updates["target_1_fraction"] = config.execution.non_a_tier_target_1_fraction + + # Per-engine overrides (highest priority) + if candidate.engine_target_1_r is not None: + execution_updates["target_1_r"] = candidate.engine_target_1_r + if candidate.engine_target_1_fraction is not None: + execution_updates["target_1_fraction"] = candidate.engine_target_1_fraction + + # Adaptive exit: adjust trailing warmup based on close_location zone + exec_cfg = config.execution + if exec_cfg.adaptive_exit_enabled: + cl = candidate.features.get("close_location") + if cl is not None: + try: + cl_val = float(cl) + except (TypeError, ValueError): + cl_val = None + if cl_val is not None: + if cl_val >= exec_cfg.adaptive_exit_exhaustion_close_min: + execution_updates["trailing_warmup_days"] = exec_cfg.adaptive_exit_exhaustion_trailing_warmup + elif exec_cfg.adaptive_exit_orderly_close_min <= cl_val <= exec_cfg.adaptive_exit_orderly_close_max: + execution_updates["trailing_warmup_days"] = exec_cfg.adaptive_exit_orderly_trailing_warmup + + if not execution_updates: + return config.execution + return config.execution.model_copy(update=execution_updates) + + +def check_next_open_gap_cap(candidate: Candidate, bar: dict[str, Any] | None) -> str | None: + """Reject next-open entries when the open gaps up more than the engine cap. + + Returns skip_reason string or None if the gap is acceptable. + Shared by BacktestRunner and PaperTradingEngine. + """ + if candidate.entry_timing_policy != "next_open": + return None + if candidate.engine_next_open_gap_cap_pct is None: + return None + if bar is None: + return None + + open_price = bar.get("open") + if open_price is None: + return None + + entry_est = candidate.entry_price_est + if entry_est is None or entry_est <= 0: + return None + + gap_pct = (float(open_price) - entry_est) / entry_est + if gap_pct > candidate.engine_next_open_gap_cap_pct: + return f"next_open_gap_too_large:{gap_pct:.2%}>{candidate.engine_next_open_gap_cap_pct:.2%}" + return None