yirugi
/
fithia2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
main
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from 'fb4fec7dac'
${ noResults }
fithia2/libs/backtest/execution.py

647 lines
24 KiB
Python
Raw Blame History

"""Fill simulation for entries and exits, plus shared execution helpers."""
from __future__ import annotations
import datetime as dt
import math
import uuid
from typing import Any
from libs.backtest.domain import (
BacktestConfig,
Candidate,
ExecutionConfig,
ExitReason,
FilledTrade,
OpenPosition,
PlannedOrder,
PositionStatus,
)
from libs.common.logging import get_logger
logger = get_logger(__name__)
# ---------------------------------------------------------------------------
# Slippage helpers
# ---------------------------------------------------------------------------
def _long_entry_fill(open_price: float, slippage_bps: float) -> float:
"""Buy at open + slippage (pays more)."""
return open_price * (1.0 + slippage_bps / 10_000)
def _long_exit_fill(price: float, slippage_bps: float) -> float:
"""Sell at price - slippage (receives less)."""
return price * (1.0 - slippage_bps / 10_000)
def _short_entry_fill(open_price: float, slippage_bps: float) -> float:
"""Sell short at open - slippage (receives less)."""
return open_price * (1.0 - slippage_bps / 10_000)
def _short_exit_fill(price: float, slippage_bps: float) -> float:
"""Buy to cover at price + slippage (pays more)."""
return price * (1.0 + slippage_bps / 10_000)
# ---------------------------------------------------------------------------
# Entry simulation
# ---------------------------------------------------------------------------
def simulate_entry(
plan: PlannedOrder,
bar: dict[str, Any],
config: ExecutionConfig,
position_id: str | None = None,
) -> OpenPosition | None:
"""Simulate filling a planned entry at the bar's open or reaction close.
Returns None (position NOT opened) if bar is missing or the required price is invalid.
No zero imputation — missing bar = no entry.
"""
if bar is None:
logger.warning("entry_skip_missing_bar", event_id=plan.candidate.event_id)
return None
if plan.skip_reason is not None:
logger.debug("entry_skip_gate_rejected", reason=plan.skip_reason)
return None
if plan.shares <= 0:
logger.warning("entry_skip_zero_shares", event_id=plan.candidate.event_id)
return None
entry_policy = plan.entry_timing_policy or "next_open"
if entry_policy == "reaction_close":
reference_price = bar.get("close")
if reference_price is None or reference_price <= 0:
logger.warning("entry_skip_invalid_close", event_id=plan.candidate.event_id, bar=bar)
return None
else:
reference_price = bar.get("open")
if reference_price is None or reference_price <= 0:
logger.warning("entry_skip_invalid_open", event_id=plan.candidate.event_id, bar=bar)
return None
is_short = plan.candidate.trade_direction == "short"
if is_short:
fill_price = _short_entry_fill(float(reference_price), config.slippage_bps_base)
else:
fill_price = _long_entry_fill(float(reference_price), config.slippage_bps_base)
slippage_bps_actual = abs(fill_price / float(reference_price) - 1.0) * 10_000
pid = position_id or str(uuid.uuid4())
# entry_date here is the bar date (execution_date of the candidate)
bar_date_raw = bar.get("date")
if isinstance(bar_date_raw, str):
entry_date = dt.date.fromisoformat(bar_date_raw)
elif isinstance(bar_date_raw, dt.date):
entry_date = bar_date_raw
else:
entry_date = plan.candidate.execution_date
return OpenPosition(
position_id=pid,
plan=plan,
entry_date=entry_date,
entry_price=fill_price,
entry_fill_slippage_bps=slippage_bps_actual,
current_stop=plan.stop_price,
target_price=plan.target_price,
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,
)
# ---------------------------------------------------------------------------
# Exit simulation
# ---------------------------------------------------------------------------
def simulate_exit(
position: OpenPosition,
bar: dict[str, Any],
config: ExecutionConfig,
current_date: dt.date,
) -> FilledTrade | None:
"""Check if position should exit on this bar. Returns FilledTrade or None.
Handles:
- Stop loss (low ≤ stop_price)
- Target (high ≥ target_price) — with partial exit support
- Same-bar conflict (controlled by same_bar_priority)
- Time exit (days_held >= max_holding_days)
- Kill switch / missing bar handled upstream
Partial exits: when target_1_fraction < 1.0 and target is hit, exits only
that fraction, moves stop to breakeven for remaining shares, and returns
the partial FilledTrade. Remaining shares continue with trailing stop.
Slippage is applied in the unfavorable direction for long positions.
"""
if bar is None:
return None
bar_low = bar.get("low")
bar_high = bar.get("high")
bar_close = bar.get("close")
slippage = config.slippage_bps_base
is_short = position.plan.candidate.trade_direction == "short"
# Direction-aware stop/target detection
if is_short:
# Short: stop is above entry (hit when price goes up), target below entry (hit when price drops)
stop_hit = bar_high is not None and float(bar_high) >= position.current_stop
target_hit = bar_low is not None and float(bar_low) <= position.target_price
exit_fill_fn = _short_exit_fill
else:
# Long: stop below entry (hit when price drops), target above entry (hit when price rises)
stop_hit = bar_low is not None and float(bar_low) <= position.current_stop
target_hit = bar_high is not None and float(bar_high) >= position.target_price
exit_fill_fn = _long_exit_fill
exit_reason: ExitReason | None = None
exit_fill_price: float | None = None
if stop_hit and target_hit:
# Same-bar conflict
if config.same_bar_priority == "stop_first_conservative":
exit_reason = ExitReason.STOP
exit_fill_price = exit_fill_fn(position.current_stop, slippage)
else: # target_first_aggressive
exit_reason = ExitReason.TARGET
exit_fill_price = exit_fill_fn(position.target_price, slippage)
elif stop_hit:
exit_reason = ExitReason.STOP
exit_fill_price = exit_fill_fn(position.current_stop, slippage)
elif target_hit:
exit_reason = ExitReason.TARGET
exit_fill_price = exit_fill_fn(position.target_price, slippage)
elif position.days_held >= config.max_holding_days:
exit_reason = ExitReason.TIME
if bar_close is not None and float(bar_close) > 0:
exit_fill_price = exit_fill_fn(float(bar_close), slippage)
else:
exit_fill_price = position.entry_price # fallback (shouldn't happen)
# No-follow-through early exit
# Long: D+1 close < entry. Short: D+1 close > entry.
if (
exit_reason is None
and config.no_follow_through_exit
and position.days_held == 1
and bar_close is not None
):
close_val = float(bar_close)
nft_triggered = (close_val > position.entry_price) if is_short else (close_val < position.entry_price)
if nft_triggered:
exit_reason = ExitReason.NO_FOLLOW_THROUGH
exit_fill_price = exit_fill_fn(close_val, slippage)
if exit_reason is None or exit_fill_price is None:
return None
# --- Partial exit logic ---
fraction = config.target_1_fraction
if (
exit_reason == ExitReason.TARGET
and fraction is not None
and 0.0 < fraction < 1.0
and position.status != PositionStatus.PARTIALLY_EXITED
):
partial_shares = max(1, math.floor(position.shares_open * fraction))
remaining_shares = position.shares_open - partial_shares
if remaining_shares > 0:
# Build partial fill trade
partial_trade = _build_filled_trade_partial(
position, exit_fill_price, exit_reason, current_date, config,
shares=partial_shares,
)
# Mutate position: reduce shares, move stop to breakeven, mark partial
position.shares_open = remaining_shares
position.current_stop = position.entry_price # breakeven stop
position.status = PositionStatus.PARTIALLY_EXITED
position.partial_fills.append(partial_trade)
return partial_trade
return _build_filled_trade(position, exit_fill_price, exit_reason, current_date, config)
def simulate_kill_switch_exit(
position: OpenPosition,
bar: dict[str, Any] | None,
current_date: dt.date,
config: ExecutionConfig,
) -> FilledTrade:
"""Force-close a position due to kill switch (portfolio drawdown)."""
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
if bar is not None and bar.get("close") is not None:
exit_price = exit_fill_fn(float(bar["close"]), slippage)
else:
exit_price = position.entry_price # last known price fallback
return _build_filled_trade(
position, exit_price, ExitReason.KILL_SWITCH, current_date, config
)
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,
config: ExecutionConfig,
) -> FilledTrade:
"""Close a position when bar data is unavailable for too long."""
return _build_filled_trade(
position, position.entry_price, ExitReason.MISSING_BAR, current_date, config
)
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
# ---------------------------------------------------------------------------
def update_trailing_stop(
position: OpenPosition,
bar: dict[str, Any],
trailing_model: str = "bar_low",
warmup_days: int = 0,
) -> None:
"""Ratchet stop towards price for trailing model. Mutates position in place.
For long: ratchet stop UP (never down). Tracks peak_price as highest high.
For short: ratchet stop DOWN (never up). Tracks peak_price as lowest low.
Models:
- "bar_low"/"bar_high": trail to bar extreme (tightest, aggressive)
- "pct_3": trail at peak_price * (1 +/- 3%) -- moderate
- "pct_5": trail at peak_price * (1 +/- 5%) -- wider
Args:
warmup_days: Skip trailing until position has been held this many days.
"""
is_short = position.plan.candidate.trade_direction == "short"
# Track best price (peak for long = highest high, for short = lowest low)
if is_short:
bar_low = bar.get("low")
if bar_low is not None:
position.peak_price = min(position.peak_price, float(bar_low))
else:
bar_high = bar.get("high")
if bar_high is not None:
position.peak_price = max(position.peak_price, float(bar_high))
# Don't tighten stop during warmup period
if position.days_held < warmup_days:
return
if is_short:
# Short: trail stop DOWN towards price (tighter = lower stop)
if trailing_model == "bar_low" or trailing_model == "bar_high":
bar_high = bar.get("high")
if bar_high is not None:
position.current_stop = min(position.current_stop, float(bar_high))
elif trailing_model.startswith("pct_"):
try:
trail_pct = float(trailing_model.split("_")[1]) / 100.0
except (IndexError, ValueError):
trail_pct = 0.03
trail_stop = position.peak_price * (1.0 + trail_pct)
position.current_stop = min(position.current_stop, trail_stop)
else:
# Long: trail stop UP (original behavior)
if trailing_model == "bar_low":
bar_low = bar.get("low")
if bar_low is not None:
new_stop = max(position.current_stop, float(bar_low))
position.current_stop = new_stop
elif trailing_model.startswith("pct_"):
try:
trail_pct = float(trailing_model.split("_")[1]) / 100.0
except (IndexError, ValueError):
trail_pct = 0.03
trail_stop = position.peak_price * (1.0 - trail_pct)
position.current_stop = max(position.current_stop, trail_stop)
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _build_filled_trade_partial(
position: OpenPosition,
exit_price: float,
exit_reason: ExitReason,
exit_date: dt.date,
config: ExecutionConfig,
shares: int,
) -> FilledTrade:
"""Build a FilledTrade for a partial exit (specific share count)."""
is_short = position.plan.candidate.trade_direction == "short"
commission = shares * config.commission_per_share # only exit leg for partial
if is_short:
gross_pnl = (position.entry_price - exit_price) * shares
else:
gross_pnl = (exit_price - position.entry_price) * shares
net_pnl = gross_pnl - commission
entry_price = position.entry_price
if is_short:
pnl_pct = (entry_price - exit_price) / entry_price if entry_price != 0 else 0.0
else:
pnl_pct = (exit_price - entry_price) / entry_price if entry_price != 0 else 0.0
stop_distance = abs(entry_price - position.plan.stop_price)
if stop_distance > 0:
r_multiple = gross_pnl / shares / stop_distance
else:
r_multiple = 0.0
holding_days = (exit_date - position.entry_date).days
trade_id = str(uuid.uuid4())
return FilledTrade(
trade_id=trade_id,
position_id=position.position_id,
event_id=position.plan.candidate.event_id,
symbol=position.plan.candidate.symbol,
event_date=position.plan.event_date or position.plan.candidate.event_date,
timing_class=position.plan.timing_class,
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,
exit_price=exit_price,
exit_reason=exit_reason,
shares=shares,
commission=commission,
slippage_bps=config.slippage_bps_base,
gross_pnl=gross_pnl,
net_pnl=net_pnl,
pnl_pct=pnl_pct,
r_multiple=r_multiple,
holding_days=holding_days,
)
def _build_filled_trade(
position: OpenPosition,
exit_price: float,
exit_reason: ExitReason,
exit_date: dt.date,
config: ExecutionConfig,
) -> FilledTrade:
is_short = position.plan.candidate.trade_direction == "short"
shares = position.shares_open
commission = shares * config.commission_per_share * 2 # entry + exit legs
if is_short:
gross_pnl = (position.entry_price - exit_price) * shares
else:
gross_pnl = (exit_price - position.entry_price) * shares
net_pnl = gross_pnl - commission
entry_price = position.entry_price
if is_short:
pnl_pct = (entry_price - exit_price) / entry_price if entry_price != 0 else 0.0
else:
pnl_pct = (exit_price - entry_price) / entry_price if entry_price != 0 else 0.0
# R-multiple uses actual fill price
stop_distance = abs(entry_price - position.plan.stop_price)
if stop_distance > 0:
r_multiple = gross_pnl / shares / stop_distance
else:
r_multiple = 0.0
holding_days = (exit_date - position.entry_date).days
trade_id = str(uuid.uuid4())
return FilledTrade(
trade_id=trade_id,
position_id=position.position_id,
event_id=position.plan.candidate.event_id,
symbol=position.plan.candidate.symbol,
event_date=position.plan.event_date or position.plan.candidate.event_date,
timing_class=position.plan.timing_class,
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,
exit_price=exit_price,
exit_reason=exit_reason,
shares=shares,
commission=commission,
slippage_bps=config.slippage_bps_base,
gross_pnl=gross_pnl,
net_pnl=net_pnl,
pnl_pct=pnl_pct,
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
Reference in New Issue View Git Blame Copy Permalink