stefan-jansen--machine-learning-for-trading
367 行
14 KiB
Python
367 行
14 KiB
Python
"""Persistent-slot signal-exit strategy for intraday execution.
|
|
|
|
The slot mechanism is the operationally-defensible variant of the canonical
|
|
``eq_w_topk`` selection used in chapters 11-19. It is designed for
|
|
microstructure case studies (nasdaq100) where:
|
|
|
|
- predictions arrive at one cadence (e.g. 1-min) but execution rebalances
|
|
on a coarser cadence (e.g. 15-min)
|
|
- per-symbol score distributions are heterogeneous (a 0.6 score for AAPL
|
|
is not equivalent to a 0.6 score for AMZN), so entry uses a per-symbol
|
|
rolling quantile rather than cross-sectional rank
|
|
- the chapter narrative needs a *signal-based exit* — close positions
|
|
when the score crosses back below a stay-threshold — which neither
|
|
``eq_w_topk`` nor ``risk_controls.position`` (Ch19) expresses
|
|
|
|
Mechanism:
|
|
1. Align predictions to price-grid timestamps via backward asof (freshest
|
|
prediction within ``pred_freshness_max_min``).
|
|
2. Compute per-symbol rolling entry threshold at quantile ``long_q``
|
|
(delegates to ``signals.per_symbol_rolling_percentile_signal``).
|
|
3. Optionally compute per-symbol rolling stay threshold at quantile
|
|
``exit_signal_q`` < ``long_q``.
|
|
4. Walk bars in time order maintaining ``open_slots: dict[sym -> entry_bar]``.
|
|
At each bar:
|
|
(a) close slots whose age >= ``hold_bars`` (max-hold backstop)
|
|
(b) close slots whose current score < stay_threshold (signal-exit)
|
|
(c) close slots hitting ``take_profit`` / ``stop_loss`` vs entry price
|
|
(d) open new slots from the top ``max_slots - len(open_slots)`` entries
|
|
sorted by score descending
|
|
(e) emit ``weight_per_slot`` for every currently-held (ts, sym)
|
|
|
|
Take-profit and stop-loss are per-slot exit legs evaluated against each
|
|
slot's entry price. They are an intrinsic property of the slot mechanism's
|
|
event-driven holding period (entry -> exit on the FIRST trigger), distinct
|
|
from Ch19 ``risk_controls.position`` overlays which act on a continuously
|
|
rebalanced weight series. The sandbox finding (nasdaq100 v4) is that
|
|
signal-exit OR take-profit each help out-of-sample but stacking them does
|
|
not, so callers sweep them as mutually exclusive exit variants.
|
|
|
|
The output schema ``[timestamp, symbol, weight]`` is what
|
|
``backtest_runner._run_engine`` consumes as ``weights``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from datetime import datetime
|
|
from typing import Literal
|
|
|
|
import polars as pl
|
|
|
|
from case_studies.utils.signals import per_symbol_rolling_percentile_signal
|
|
|
|
|
|
def _run_slot_simulation(
|
|
signals_by_ts: dict[datetime, list[tuple[str, float]]],
|
|
all_bars_sorted: list[datetime],
|
|
max_slots: int,
|
|
weight_per_slot: float,
|
|
hold_bars: int,
|
|
*,
|
|
score_by_ts_sym: Mapping[tuple[datetime, str], float] | None,
|
|
stay_threshold_by_ts_sym: Mapping[tuple[datetime, str], float] | None,
|
|
price_by_ts_sym: Mapping[tuple[datetime, str], float] | None = None,
|
|
take_profit: float | None = None,
|
|
stop_loss: float | None = None,
|
|
) -> tuple[pl.DataFrame, dict]:
|
|
"""Pure-mechanism slot simulator with optional signal-exit and TP/SL.
|
|
|
|
Walks ``all_bars_sorted`` in order. Returns long-only weights frame and
|
|
a stats dict with per-exit-cause counts. Exit priority per bar:
|
|
max-hold, then signal-exit, then take-profit, then stop-loss. TP/SL
|
|
compare the current bar's price to the slot's entry price and require
|
|
``price_by_ts_sym``; absent a current/entry price the TP/SL legs are
|
|
skipped (the slot still honours max-hold/signal-exit).
|
|
"""
|
|
if max_slots <= 0:
|
|
raise ValueError(f"max_slots must be positive, got {max_slots}")
|
|
if hold_bars <= 0:
|
|
raise ValueError(f"hold_bars must be positive, got {hold_bars}")
|
|
if not (0 < weight_per_slot <= 1.0):
|
|
raise ValueError(f"weight_per_slot must be in (0, 1], got {weight_per_slot}")
|
|
if stop_loss is not None and stop_loss < 0:
|
|
raise ValueError(f"stop_loss must be positive (sign applied internally), got {stop_loss}")
|
|
if take_profit is not None and take_profit <= 0:
|
|
raise ValueError(f"take_profit must be positive, got {take_profit}")
|
|
|
|
score_lookup = score_by_ts_sym or {}
|
|
stay_lookup = stay_threshold_by_ts_sym or {}
|
|
price_lookup = price_by_ts_sym or {}
|
|
use_tp_sl = take_profit is not None or stop_loss is not None
|
|
|
|
open_slots: dict[str, dict] = {} # sym -> {"entry_i", "entry_px"}
|
|
rows: list[dict] = []
|
|
n_entries = 0
|
|
n_exits_maxhold = 0
|
|
n_exits_signal = 0
|
|
n_exits_tp = 0
|
|
n_exits_sl = 0
|
|
|
|
for i, ts in enumerate(all_bars_sorted):
|
|
# 1. Expire slots — max-hold, then signal-exit, then TP, then SL
|
|
to_close: list[tuple[str, str]] = []
|
|
for sym, slot in open_slots.items():
|
|
if i - slot["entry_i"] >= hold_bars:
|
|
to_close.append((sym, "maxhold"))
|
|
continue
|
|
key = (ts, sym)
|
|
current_score = score_lookup.get(key)
|
|
stay_thresh = stay_lookup.get(key)
|
|
if (
|
|
current_score is not None
|
|
and stay_thresh is not None
|
|
and current_score < stay_thresh
|
|
):
|
|
to_close.append((sym, "signal"))
|
|
continue
|
|
if use_tp_sl and slot["entry_px"] is not None:
|
|
current_px = price_lookup.get(key)
|
|
if current_px is not None and slot["entry_px"] > 0:
|
|
ret = current_px / slot["entry_px"] - 1.0
|
|
if take_profit is not None and ret >= take_profit:
|
|
to_close.append((sym, "tp"))
|
|
continue
|
|
if stop_loss is not None and ret <= -stop_loss:
|
|
to_close.append((sym, "sl"))
|
|
continue
|
|
|
|
for sym, cause in to_close:
|
|
del open_slots[sym]
|
|
if cause == "maxhold":
|
|
n_exits_maxhold += 1
|
|
elif cause == "signal":
|
|
n_exits_signal += 1
|
|
elif cause == "tp":
|
|
n_exits_tp += 1
|
|
else:
|
|
n_exits_sl += 1
|
|
|
|
# 2. New entries — sorted by score desc, capacity-limited
|
|
candidates = signals_by_ts.get(ts, [])
|
|
if candidates:
|
|
fresh = [(s, sc) for s, sc in candidates if s not in open_slots]
|
|
fresh.sort(key=lambda x: -x[1])
|
|
capacity = max_slots - len(open_slots)
|
|
for sym, _score in fresh[:capacity]:
|
|
entry_px = price_lookup.get((ts, sym)) if use_tp_sl else None
|
|
open_slots[sym] = {"entry_i": i, "entry_px": entry_px}
|
|
n_entries += 1
|
|
|
|
# 3. Emit weights for currently-held symbols
|
|
for sym in open_slots:
|
|
rows.append({"timestamp": ts, "symbol": sym, "weight": weight_per_slot})
|
|
|
|
stats = {
|
|
"n_entries": n_entries,
|
|
"n_exits_maxhold": n_exits_maxhold,
|
|
"n_exits_signal": n_exits_signal,
|
|
"n_exits_tp": n_exits_tp,
|
|
"n_exits_sl": n_exits_sl,
|
|
"n_exits_total": n_exits_maxhold + n_exits_signal + n_exits_tp + n_exits_sl,
|
|
"max_slots": max_slots,
|
|
"hold_bars": hold_bars,
|
|
"n_bars": len(all_bars_sorted),
|
|
}
|
|
if not rows:
|
|
empty = pl.DataFrame(
|
|
schema={"timestamp": pl.Datetime("us"), "symbol": pl.String, "weight": pl.Float64}
|
|
)
|
|
return empty, stats
|
|
out = pl.DataFrame(rows).with_columns(pl.col("timestamp").cast(pl.Datetime("us")))
|
|
return out, stats
|
|
|
|
|
|
def _align_predictions_to_bars(
|
|
predictions: pl.DataFrame,
|
|
bar_grid: pl.DataFrame,
|
|
*,
|
|
pred_freshness_max_min: int | None,
|
|
score_col: str,
|
|
time_col: str,
|
|
asset_col: str,
|
|
) -> pl.DataFrame:
|
|
"""Backward-asof align predictions to a (symbol, timestamp) bar grid.
|
|
|
|
``bar_grid`` carries the rebalance schedule. For each (sym, bar_ts) row,
|
|
pull the freshest prediction with timestamp <= bar_ts and stale by at
|
|
most ``pred_freshness_max_min`` minutes. Predictions older than the
|
|
tolerance are dropped, leaving rows with null ``y_score`` which are
|
|
then filtered out.
|
|
|
|
When ``pred_freshness_max_min`` is None, the asof tolerance is
|
|
unbounded (typical when predictions and prices share the same cadence).
|
|
"""
|
|
bars = bar_grid.select([asset_col, time_col]).sort([asset_col, time_col])
|
|
preds = predictions.select([asset_col, time_col, score_col]).sort([asset_col, time_col])
|
|
tol = f"{pred_freshness_max_min}m" if pred_freshness_max_min is not None else None
|
|
aligned = bars.join_asof(
|
|
preds,
|
|
on=time_col,
|
|
by=asset_col,
|
|
strategy="backward",
|
|
tolerance=tol,
|
|
)
|
|
return aligned.filter(pl.col(score_col).is_not_null()).sort([asset_col, time_col])
|
|
|
|
|
|
def _signals_to_lookup(
|
|
signals_df: pl.DataFrame,
|
|
*,
|
|
score_col: str,
|
|
time_col: str,
|
|
asset_col: str,
|
|
) -> dict[datetime, list[tuple[str, float]]]:
|
|
"""Convert ``per_symbol_rolling_percentile_signal`` output (signal==1 rows)
|
|
to ``dict[ts -> list[(sym, score)]]`` for the slot simulator.
|
|
"""
|
|
fired = signals_df.filter(pl.col("signal") == 1).select([time_col, asset_col, score_col])
|
|
out: dict[datetime, list[tuple[str, float]]] = {}
|
|
for row in fired.iter_rows(named=True):
|
|
ts = row[time_col]
|
|
if ts not in out:
|
|
out[ts] = []
|
|
out[ts].append((row[asset_col], float(row[score_col])))
|
|
return out
|
|
|
|
|
|
def build_persistent_slot_weights_hybrid(
|
|
predictions: pl.DataFrame,
|
|
prices: pl.DataFrame,
|
|
*,
|
|
long_q: float,
|
|
lookback_days: int,
|
|
bars_per_day: int,
|
|
max_slots: int,
|
|
hold_bars: int,
|
|
weight_per_slot: float | None = None,
|
|
exit_signal_q: float | None = None,
|
|
take_profit: float | None = None,
|
|
stop_loss: float | None = None,
|
|
pred_freshness_max_min: int | None = None,
|
|
direction: Literal["long_only", "short_only"] = "long_only",
|
|
score_col: str = "y_score",
|
|
time_col: str = "timestamp",
|
|
asset_col: str = "symbol",
|
|
price_col: str = "close",
|
|
) -> tuple[pl.DataFrame, dict]:
|
|
"""Library entry point for the persistent-slot signal-exit selection method.
|
|
|
|
Pipeline:
|
|
1. Align ``predictions`` to ``prices`` grid via backward-asof.
|
|
2. Compute per-symbol rolling entry threshold at ``long_q``; entry signal
|
|
where y_score > threshold.
|
|
3. If ``exit_signal_q`` is set (< ``long_q``), compute per-symbol rolling
|
|
stay threshold at that quantile.
|
|
4. Run slot simulation with ``max_slots`` capacity and ``hold_bars`` cap,
|
|
plus optional ``take_profit`` / ``stop_loss`` per-slot exit legs
|
|
(evaluated on the bar-close return vs the slot's entry price).
|
|
5. Apply ``direction`` sign — ``short_only`` flips the weight sign.
|
|
|
|
``take_profit`` / ``stop_loss`` are decimals (0.005 = 0.5%). They free the
|
|
slot on trigger so the weight series stops emitting the symbol until a fresh
|
|
entry signal — the slot-native re-entry semantics that engine-level
|
|
``risk_controls.position`` rules cannot express against a dense target
|
|
series. The nasdaq100 v4 sandbox finding is that signal-exit OR take-profit
|
|
each help out-of-sample but stacking them does not, so callers pass at most
|
|
one of ``exit_signal_q`` / ``take_profit`` per configuration.
|
|
|
|
Returns ``(weights_df, stats_dict)``. ``weights_df`` has schema
|
|
``[timestamp, symbol, weight]`` matching ``_run_engine`` input.
|
|
|
|
Note: long_short is not supported — slot books are inherently
|
|
single-direction (a symbol cannot occupy a long and short slot
|
|
simultaneously). The cross-asset long-short story belongs to
|
|
``eq_w_topk`` / ``quintile_long_short``.
|
|
"""
|
|
if direction not in ("long_only", "short_only"):
|
|
raise ValueError(
|
|
f"slot direction must be 'long_only' or 'short_only', got {direction!r}; "
|
|
"long_short is not supported for the slot mechanism"
|
|
)
|
|
if exit_signal_q is not None and exit_signal_q >= long_q:
|
|
raise ValueError(
|
|
f"exit_signal_q ({exit_signal_q}) must be < long_q ({long_q}) "
|
|
"so the stay threshold sits below the entry threshold"
|
|
)
|
|
if weight_per_slot is None:
|
|
weight_per_slot = 1.0 / max_slots
|
|
|
|
aligned = _align_predictions_to_bars(
|
|
predictions,
|
|
prices,
|
|
pred_freshness_max_min=pred_freshness_max_min,
|
|
score_col=score_col,
|
|
time_col=time_col,
|
|
asset_col=asset_col,
|
|
)
|
|
|
|
sig_df = per_symbol_rolling_percentile_signal(
|
|
aligned,
|
|
long_q=long_q,
|
|
lookback_days=lookback_days,
|
|
bars_per_day=bars_per_day,
|
|
score_col=score_col,
|
|
time_col=time_col,
|
|
asset_col=asset_col,
|
|
signal_type="long_only",
|
|
stay_q=exit_signal_q,
|
|
)
|
|
|
|
signals_by_ts = _signals_to_lookup(
|
|
sig_df,
|
|
score_col=score_col,
|
|
time_col=time_col,
|
|
asset_col=asset_col,
|
|
)
|
|
|
|
if exit_signal_q is not None:
|
|
stay_lookup = {
|
|
(r[time_col], r[asset_col]): float(r["stay_thresh"])
|
|
for r in sig_df.filter(pl.col("stay_thresh").is_not_null())
|
|
.select([time_col, asset_col, "stay_thresh"])
|
|
.iter_rows(named=True)
|
|
}
|
|
score_lookup = {
|
|
(r[time_col], r[asset_col]): float(r[score_col])
|
|
for r in aligned.select([time_col, asset_col, score_col]).iter_rows(named=True)
|
|
}
|
|
else:
|
|
stay_lookup = None
|
|
score_lookup = None
|
|
|
|
if take_profit is not None or stop_loss is not None:
|
|
price_lookup = {
|
|
(r[time_col], r[asset_col]): float(r[price_col])
|
|
for r in prices.select([time_col, asset_col, price_col])
|
|
.filter(pl.col(price_col).is_not_null())
|
|
.iter_rows(named=True)
|
|
}
|
|
else:
|
|
price_lookup = None
|
|
|
|
schedule = sorted(aligned[time_col].unique().to_list())
|
|
|
|
weights, stats = _run_slot_simulation(
|
|
signals_by_ts=signals_by_ts,
|
|
all_bars_sorted=schedule,
|
|
max_slots=max_slots,
|
|
weight_per_slot=weight_per_slot,
|
|
hold_bars=hold_bars,
|
|
score_by_ts_sym=score_lookup,
|
|
stay_threshold_by_ts_sym=stay_lookup,
|
|
price_by_ts_sym=price_lookup,
|
|
take_profit=take_profit,
|
|
stop_loss=stop_loss,
|
|
)
|
|
|
|
if direction == "short_only" and not weights.is_empty():
|
|
weights = weights.with_columns((-pl.col("weight")).alias("weight"))
|
|
|
|
stats["direction"] = direction
|
|
stats["long_q"] = long_q
|
|
stats["exit_signal_q"] = exit_signal_q
|
|
stats["take_profit"] = take_profit
|
|
stats["stop_loss"] = stop_loss
|
|
return weights, stats
|