File size: 7,426 Bytes
d53dc44 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 | """
ml/execution.py — Advanced execution optimization.
Capabilities:
1. Spread spike avoidance (skip entry when spread > threshold)
2. Refined time-of-day filters with intraday session scoring
3. Volume-weighted entry timing
4. Microstructure filters (tick-level proxies)
5. Partial fill simulation with realistic fill rates
6. Latency modeling (configurable delay)
7. Pullback confirmation entries (wait for better price)
"""
from __future__ import annotations
import logging
import numpy as np
import pandas as pd
logger = logging.getLogger("trading_system.ml.execution")
class ExecutionOptimizer:
"""Execution quality improvements.
Filters out entries during:
- First/last 15 minutes (spread spikes)
- Low-liquidity windows (volume < 30% of session avg)
- High-impact macro windows (configurable)
- Microstructure anomalies (extreme range, thin volume)
"""
def __init__(
self,
skip_first_min: int = 15,
skip_last_min: int = 120, # stop 2h before close (2:00 PM ET)
min_volume_ratio: float = 0.30,
max_spread_atr_ratio: float = 0.5,
latency_bars: int = 1,
partial_fill_rate: float = 0.95,
pullback_atr_frac: float = 0.2,
pullback_max_wait: int = 3,
):
self.skip_first_min = skip_first_min
self.skip_last_min = skip_last_min
self.min_volume_ratio = min_volume_ratio
self.max_spread_atr_ratio = max_spread_atr_ratio
self.latency_bars = latency_bars
self.partial_fill_rate = partial_fill_rate
self.pullback_atr_frac = pullback_atr_frac
self.pullback_max_wait = pullback_max_wait
def compute_execution_mask(
self,
df: pd.DataFrame,
atr: pd.Series,
) -> pd.Series:
"""Return boolean mask: True = OK to enter, False = skip.
Args:
df: OHLCV DataFrame with DatetimeIndex.
atr: ATR series aligned to df.
Returns: pd.Series[bool] indexed like df.
"""
ok = pd.Series(True, index=df.index)
# Time-of-day filter
if df.index.tzinfo is not None:
et_index = df.index.tz_convert("US/Eastern")
else:
try:
et_index = df.index.tz_localize("UTC").tz_convert("US/Eastern")
except Exception:
et_index = df.index
et_minutes = et_index.hour * 60 + et_index.minute
market_open = 9 * 60 + 30 # 9:30 ET
market_close = 16 * 60 # 16:00 ET
# Skip first N minutes
ok &= et_minutes >= (market_open + self.skip_first_min)
# Skip last N minutes
ok &= et_minutes <= (market_close - self.skip_last_min)
# Volume filter: skip low-liquidity bars
volume = df["volume"].astype(float)
vol_session_avg = volume.rolling(78).mean() # ~1 day on 5m bars
vol_ratio = volume / vol_session_avg.replace(0, np.nan)
ok &= vol_ratio.fillna(1.0) >= self.min_volume_ratio
# Spread proxy: (high - low) / ATR
bar_range = (df["high"].astype(float) - df["low"].astype(float))
spread_ratio = bar_range / atr.replace(0, np.nan)
ok &= spread_ratio.fillna(0) <= self.max_spread_atr_ratio * 3
# Microstructure: skip if bar volume is extremely thin (< 100 shares)
ok &= volume >= 100
return ok
def compute_session_quality(self, df: pd.DataFrame) -> pd.Series:
"""Score each bar's execution quality (0-1) based on time-of-day pattern.
Higher scores during mid-session liquid periods, lower near open/close.
"""
if df.index.tzinfo is not None:
et_index = df.index.tz_convert("US/Eastern")
else:
try:
et_index = df.index.tz_localize("UTC").tz_convert("US/Eastern")
except Exception:
et_index = df.index
minutes = et_index.hour * 60 + et_index.minute
market_open = 9 * 60 + 30
market_close = 16 * 60
# Minutes into session
session_min = (minutes - market_open).values.astype(float)
session_len = market_close - market_open # 390
# Parabolic quality curve: best in middle, worst at edges
x = np.clip(session_min / session_len, 0, 1)
quality = 4 * x * (1 - x) # peak 1.0 at midday
return pd.Series(quality, index=df.index).clip(0, 1)
def apply_latency(self, signal_idx: int, max_idx: int) -> int:
"""Delay signal execution by latency_bars to simulate real-world latency."""
return min(signal_idx + self.latency_bars, max_idx)
def simulate_partial_fill(
self,
desired_shares: float,
bar_volume: float,
participation_rate: float = 0.02,
) -> float:
"""Simulate partial fills based on bar volume.
Args:
desired_shares: Requested position size in shares.
bar_volume: Volume of the entry bar.
participation_rate: Max fraction of bar volume we'll take.
Returns:
Filled shares (may be less than desired).
"""
max_from_volume = bar_volume * participation_rate
fillable = min(desired_shares, max_from_volume)
return fillable * self.partial_fill_rate
def pullback_entry_price(
self,
side: str,
signal_price: float,
subsequent_lows: np.ndarray,
subsequent_highs: np.ndarray,
atr: float,
) -> tuple[float, int] | None:
"""Attempt a pullback entry within max_wait bars.
For buys: wait for price to dip below signal_price - pullback_frac * ATR.
For sells: wait for price to rise above signal_price + pullback_frac * ATR.
Returns:
(entry_price, bars_waited) or None if pullback never triggered.
"""
target_offset = self.pullback_atr_frac * atr
max_bars = min(self.pullback_max_wait, len(subsequent_lows))
for i in range(max_bars):
if side == "buy":
if subsequent_lows[i] <= signal_price - target_offset:
return signal_price - target_offset, i + 1
else:
if subsequent_highs[i] >= signal_price + target_offset:
return signal_price + target_offset, i + 1
return None # Pullback didn't happen, use market entry
def adjust_entry_price(
self,
side: str,
open_price: float,
high: float,
low: float,
atr: float,
) -> float:
"""Apply realistic slippage model based on bar characteristics.
For buys: entry slightly above open (adverse fill)
For sells: entry slightly below open (adverse fill)
Slippage scales with volatility.
"""
base_slip_pct = 0.02 / 100
bar_range = high - low
vol_slip = 0.0
if atr > 0:
vol_factor = min(bar_range / atr, 2.0)
vol_slip = open_price * base_slip_pct * vol_factor
slippage = open_price * base_slip_pct + vol_slip
if side == "buy":
return open_price + slippage
else:
return open_price - slippage
|