Convertendo pontuações de modelos em sinais e pesos de carteira
Código Machine Learning for Trading
Resumo
Este documento descreve várias formas de converter pontuações de previsão em sinais de trading: limites fixos, percentis móveis por ativo e percentis transversais. Limites fixos usam um nível de pontuação escolhido; limites móveis se adaptam à distribuição recente de pontuações de um ativo; limites transversais selecionam ativos em relação aos pares em cada instante. Cada abordagem pode gerar sinais apenas comprados ou comprados e vendidos, com limites inferiores simétricos usados para entradas vendidas.
O documento também apresenta métodos de ponderação de carteira, incluindo pesos iguais e pesos proporcionais à pontuação entre os ativos mais bem classificados, além da seleção comprada e vendida dos grupos de topo e de base. Quando a soma das pontuações selecionadas é zero, a ponderação por pontuação volta a usar pesos iguais. A opção de volatilidade inversa é explicitamente um espaço reservado que requer retornos históricos. O trecho também mostra um seletor de configuração que encaminha as escolhas ao método correspondente. Essas são descrições de implementação, não evidências de desempenho: o documento não apresenta resultados de backtest, custos de transação nem validação da qualidade dos sinais; portanto, os métodos, por si só, não estabelecem uma estratégia eficaz.
Ideias principais
- Limites fixos convertem pontuações em entradas compradas ou compradas e vendidas usando níveis absolutos de pontuação.
- Percentis móveis adaptam os limites de entrada e saída à distribuição recente de pontuações de cada ativo.
- Percentis transversais selecionam ativos conforme suas pontuações relativas em cada instante.
- Ativos mais bem classificados podem receber pesos iguais ou proporcionais à pontuação; a seleção comprada e vendida mantém os lados separados.
- A ponderação por volatilidade inversa é apenas um espaço reservado e requer dados históricos de retorno.
Tags
Texto completo
# signals.py
```py
"""Signal conversion and weight construction for backtesting.
Converts ML model predictions (probabilities or scores) into trading signals
and portfolio weights. Used by Ch16-20 notebooks.
Signal conversion approaches:
1. Fixed threshold: signal when score exceeds cutoff
2. Rolling percentile: signal when score exceeds recent distribution quantile
3. Cross-sectional percentile: signal for top-N% of assets at each rebalance
Weight construction:
- Equal-weight top-K: rank and select, uniform allocation
- Score-weighted top-K: rank and select, weight proportional to score
"""
from __future__ import annotations
from typing import Literal
import polars as pl
# ---------------------------------------------------------------------------
# Signal conversion
# ---------------------------------------------------------------------------
def fixed_threshold_signal(
predictions: pl.DataFrame,
threshold: float = 0.5,
score_col: str = "y_score",
signal_type: Literal["long_only", "long_short"] = "long_only",
) -> pl.DataFrame:
"""Convert predictions to signals using a fixed threshold.
For classification predictions, the score is typically a probability [0, 1].
For regression predictions, the score may need normalization first.
Args:
predictions: DataFrame with at least [timestamp, symbol, y_score]
threshold: Score threshold for entry signal
score_col: Column containing prediction scores
signal_type: "long_only" (signal=1 when above threshold) or
"long_short" (signal=1 above, signal=-1 below mirror threshold)
Returns:
DataFrame with added 'signal' column (-1, 0, or 1)
"""
if signal_type == "long_only":
return predictions.with_columns(
signal=pl.when(pl.col(score_col) > threshold).then(1).otherwise(0).cast(pl.Int8)
)
else: # long_short
lower_threshold = 1.0 - threshold
return predictions.with_columns(
signal=pl.when(pl.col(score_col) > threshold)
.then(1)
.when(pl.col(score_col) < lower_threshold)
.then(-1)
.otherwise(0)
.cast(pl.Int8)
)
def rolling_percentile_signal(
predictions: pl.DataFrame,
window: int = 63,
percentile: float = 90.0,
score_col: str = "y_score",
time_col: str = "timestamp",
asset_col: str = "symbol",
signal_type: Literal["long_only", "long_short"] = "long_only",
) -> pl.DataFrame:
"""Convert predictions to signals using rolling percentile threshold.
Computes a rolling percentile of recent scores per asset and generates
entry signals when the current score exceeds this adaptive threshold.
Args:
predictions: DataFrame with at least [timestamp, symbol, y_score]
window: Rolling window size (e.g., 63 for ~3 months of daily data)
percentile: Percentile threshold (e.g., 90 for top 10%)
score_col: Column containing prediction scores
time_col: Column containing timestamps
asset_col: Column containing asset identifiers
signal_type: "long_only" or "long_short"
Returns:
DataFrame with added 'signal' column and 'rolling_threshold' column
"""
df = predictions.sort(time_col)
df = df.with_columns(
rolling_threshold=pl.col(score_col)
.rolling_quantile(quantile=percentile / 100.0, window_size=window)
.over(asset_col)
)
if signal_type == "long_only":
df = df.with_columns(
signal=pl.when(pl.col(score_col) > pl.col("rolling_threshold"))
.then(1)
.otherwise(0)
.cast(pl.Int8)
)
else: # long_short
lower_percentile = 100.0 - percentile
df = df.with_columns(
rolling_lower_threshold=pl.col(score_col)
.rolling_quantile(quantile=lower_percentile / 100.0, window_size=window)
.over(asset_col)
)
df = df.with_columns(
signal=pl.when(pl.col(score_col) > pl.col("rolling_threshold"))
.then(1)
.when(pl.col(score_col) < pl.col("rolling_lower_threshold"))
.then(-1)
.otherwise(0)
.cast(pl.Int8)
)
return df
def cross_sectional_percentile_signal(
predictions: pl.DataFrame,
percentile: float = 90.0,
score_col: str = "y_score",
time_col: str = "timestamp",
signal_type: Literal["long_only", "long_short"] = "long_only",
) -> pl.DataFrame:
"""Convert predictions to signals using cross-sectional percentile.
At each timestamp, selects assets in the top N% by score. Controls
position count regardless of absolute score levels.
Args:
predictions: DataFrame with at least [timestamp, symbol, y_score]
percentile: Percentile cutoff (e.g., 90 for top 10% of assets)
score_col: Column containing prediction scores
time_col: Column containing timestamps
signal_type: "long_only" or "long_short"
Returns:
DataFrame with added 'signal' column and 'cs_threshold' column
"""
df = predictions.with_columns(
cs_threshold=pl.col(score_col).quantile(percentile / 100.0).over(time_col)
)
if signal_type == "long_only":
df = df.with_columns(
signal=pl.when(pl.col(score_col) >= pl.col("cs_threshold"))
.then(1)
.otherwise(0)
.cast(pl.Int8)
)
else: # long_short
lower_percentile = 100.0 - percentile
df = df.with_columns(
cs_lower_threshold=pl.col(score_col).quantile(lower_percentile / 100.0).over(time_col)
)
df = df.with_columns(
signal=pl.when(pl.col(score_col) >= pl.col("cs_threshold"))
.then(1)
.when(pl.col(score_col) <= pl.col("cs_lower_threshold"))
.then(-1)
.otherwise(0)
.cast(pl.Int8)
)
return df
# ---------------------------------------------------------------------------
# Weight construction
# ---------------------------------------------------------------------------
def build_target_weights(
predictions: pl.DataFrame,
method: Literal[
"equal_weight_top_k",
"score_weighted_top_k",
"inverse_vol",
] = "equal_weight_top_k",
top_k: int = 10,
long_short: bool = False,
score_col: str = "y_score",
time_col: str = "timestamp",
asset_col: str = "symbol",
) -> pl.DataFrame:
"""Convert predictions to portfolio target weights.
Args:
predictions: DataFrame with at least [timestamp, asset, y_score]
method: Weight construction method
top_k: Number of assets to select per rebalance
long_short: If True, go long top_k and short bottom_k
score_col: Column with prediction scores
time_col: Timestamp column
asset_col: Asset identifier column
Returns:
DataFrame with [timestamp, asset, weight] — weights sum to ~1.0 per timestamp
"""
df = predictions.sort(time_col)
# Rank assets cross-sectionally at each timestamp
df = df.with_columns(
cs_rank=pl.col(score_col).rank(method="ordinal", descending=True).over(time_col),
n_assets=pl.col(score_col).count().over(time_col),
)
# Long and short selections must remain disjoint. With an odd universe,
# leave the median asset unselected when top_k reaches half the panel.
max_side = pl.col("n_assets") // 2 if long_short else pl.col("n_assets")
df = df.with_columns(
eff_k=pl.min_horizontal(pl.lit(top_k), max_side),
)
if method == "equal_weight_top_k":
if long_short:
df = df.with_columns(
weight=pl.when(pl.col("cs_rank") <= pl.col("eff_k"))
.then(1.0 / pl.col("eff_k"))
.when(pl.col("cs_rank") > pl.col("n_assets") - pl.col("eff_k"))
.then(-1.0 / pl.col("eff_k"))
.otherwise(0.0)
)
else:
df = df.with_columns(
weight=pl.when(pl.col("cs_rank") <= pl.col("eff_k"))
.then(1.0 / pl.col("eff_k"))
.otherwise(0.0)
)
elif method == "score_weighted_top_k":
# When the top-K absolute-score sum is 0 at a timestamp (all top-K
# predictions exactly zero), score-proportional weighting would
# divide by zero. Fall back to equal-weight within the top-K for
# those timestamps so the rebalance is well-defined.
if long_short:
top = df.filter(pl.col("cs_rank") <= pl.col("eff_k"))
bottom = df.filter(pl.col("cs_rank") > pl.col("n_assets") - pl.col("eff_k"))
top_denom = pl.col(score_col).abs().sum().over(time_col)
top = top.with_columns(
weight=pl.when(top_denom > 0)
.then(pl.col(score_col).abs() / top_denom)
.otherwise(1.0 / pl.col("eff_k"))
)
bottom_denom = pl.col(score_col).abs().sum().over(time_col)
bottom = bottom.with_columns(
weight=pl.when(bottom_denom > 0)
.then(-pl.col(score_col).abs() / bottom_denom)
.otherwise(-1.0 / pl.col("eff_k"))
)
mid = df.filter(
(pl.col("cs_rank") > pl.col("eff_k"))
& (pl.col("cs_rank") <= pl.col("n_assets") - pl.col("eff_k"))
).with_columns(weight=pl.lit(0.0))
df = pl.concat([top, mid, bottom], how="diagonal_relaxed")
else:
top = df.filter(pl.col("cs_rank") <= pl.col("eff_k"))
top_denom = pl.col(score_col).abs().sum().over(time_col)
top = top.with_columns(
weight=pl.when(top_denom > 0)
.then(pl.col(score_col).abs() / top_denom)
.otherwise(1.0 / pl.col("eff_k"))
)
rest = df.filter(pl.col("cs_rank") > pl.col("eff_k")).with_columns(weight=pl.lit(0.0))
df = pl.concat([top, rest], how="diagonal_relaxed")
elif method == "inverse_vol":
# Placeholder — requires historical returns; full impl in Ch17
df = df.with_columns(
weight=pl.when(pl.col("cs_rank") <= pl.col("eff_k"))
.then(1.0 / pl.col("eff_k"))
.otherwise(0.0)
)
# Clean up helper columns
result = df.select([time_col, asset_col, "weight"]).filter(pl.col("weight") != 0.0)
return result.sort(time_col, asset_col)
# ---------------------------------------------------------------------------
# Config-driven dispatcher
# ---------------------------------------------------------------------------
def _signals_to_equal_weights(
df: pl.DataFrame,
time_col: str = "timestamp",
asset_col: str = "symbol",
) -> pl.DataFrame:
"""Convert a signal column ({-1, 0, 1}) to equal weights within each group.
Long signals get +1/N_long, short signals get -1/N_short, zero signals excluded.
"""
# Count longs and shorts per timestamp
df = df.with_columns(
n_long=pl.col("signal").filter(pl.col("signal") > 0).count().over(time_col),
n_short=pl.col("signal").filter(pl.col("signal") < 0).count().over(time_col),
)
df = df.with_columns(
weight=pl.when(pl.col("signal") > 0)
.then(1.0 / pl.col("n_long"))
.when(pl.col("signal") < 0)
.then(-1.0 / pl.col("n_short"))
.otherwise(0.0)
)
return (
df.select([time_col, asset_col, "weight"])
.filter(pl.col("weight") != 0.0)
.sort(time_col, asset_col)
)
def per_symbol_rolling_percentile_signal(
predictions: pl.DataFrame,
long_q: float = 0.80,
lookback_days: int = 20,
bars_per_day: int = 390,
score_col: str = "y_score",
time_col: str = "timestamp",
asset_col: str = "symbol",
signal_type: Literal["long_only", "long_short"] = "long_only",
stay_q: float | None = None,
) -> pl.DataFrame:
"""Per-symbol time-series rolling-percentile entry — ranks within own history.
At the first bar of each session, computes the trailing rolling quantile
of `y_score` over the past `lookback_days × bars_per_day` rows per symbol
(causal: shifted by 1 before rolling). The day's threshold is held
constant for that session's later bars via forward-fill within (symbol, date).
A bar enters when `y_score` crosses its symbol-specific session threshold:
long_only: signal=+1 if y_score > p_long
long_short: signal=+1 if y_score > p_long, -1 if y_score < (1 - long_q)
The short tail is the symmetric complement of `long_q` (so `long_q=0.85`
sets `p_short` at the 0.15 quantile). The function asserts `long_q > 0.5`
in `long_short` mode to keep `p_long > p_short`.
Warm-up: `min_samples=W // 2` means roughly the first `lookback_days / 2`
sessions per symbol have a null rolling quantile and therefore a null
threshold; the signal is coerced to 0 for those bars (no entry).
When ``stay_q`` is provided (must be < ``long_q``), a second rolling
quantile is computed at that lower level using identical windowing and
daily anchoring, exposed as a ``stay_thresh`` column on the output. The
stay threshold is used by ``slot_strategy.build_persistent_slot_weights_hybrid``
for signal-based slot exits; the entry signal column is unchanged.
Mirrors the polars aggregator at
`agents/.agents/work/nasdaq100_v3/scripts/sweep_daily_thresh_v2.py::add_daily_pct`,
which is the canonical reference for the nasdaq100 v3 strategy.
"""
if signal_type == "long_short" and long_q <= 0.5:
msg = (
f"per_symbol_rolling_percentile_signal long_short requires long_q > 0.5 "
f"to keep p_long > p_short; got long_q={long_q}"
)
raise ValueError(msg)
if stay_q is not None and stay_q >= long_q:
msg = (
f"per_symbol_rolling_percentile_signal stay_q must be < long_q so the "
f"stay threshold sits below the entry threshold; got stay_q={stay_q}, "
f"long_q={long_q}"
)
raise ValueError(msg)
W = int(lookback_days * bars_per_day)
df = predictions.sort([asset_col, time_col]).with_columns(
_date=pl.col(time_col).dt.date(),
)
y_lag = pl.col(score_col).shift(1).over(asset_col)
df = df.with_columns(
_raw_p_long=y_lag.rolling_quantile(
quantile=long_q,
window_size=W,
min_samples=W // 2,
).over(asset_col),
)
if signal_type == "long_short":
df = df.with_columns(
_raw_p_short=y_lag.rolling_quantile(
quantile=1 - long_q,
window_size=W,
min_samples=W // 2,
).over(asset_col),
)
if stay_q is not None:
df = df.with_columns(
_raw_p_stay=y_lag.rolling_quantile(
quantile=stay_q,
window_size=W,
min_samples=W // 2,
).over(asset_col),
)
df = (
df.with_columns(
_is_first=(pl.col(time_col) == pl.col(time_col).min().over([asset_col, "_date"])),
)
.with_columns(
_p_long_seed=pl.when(pl.col("_is_first")).then(pl.col("_raw_p_long")).otherwise(None),
)
.with_columns(
p_long=pl.col("_p_long_seed").forward_fill().over([asset_col, "_date"]),
)
)
if signal_type == "long_short":
df = (
df.with_columns(
_p_short_seed=pl.when(pl.col("_is_first"))
.then(pl.col("_raw_p_short"))
.otherwise(None),
)
.with_columns(
p_short=pl.col("_p_short_seed").forward_fill().over([asset_col, "_date"]),
)
.with_columns(
signal=pl.when(
pl.col("p_long").is_not_null() & (pl.col(score_col) > pl.col("p_long"))
)
.then(pl.lit(1).cast(pl.Int8))
.when(pl.col("p_short").is_not_null() & (pl.col(score_col) < pl.col("p_short")))
.then(pl.lit(-1).cast(pl.Int8))
.otherwise(pl.lit(0).cast(pl.Int8))
)
)
drop_cols = [
"_date",
"_raw_p_long",
"_raw_p_short",
"_is_first",
"_p_long_seed",
"_p_short_seed",
"p_long",
"p_short",
]
else:
df = df.with_columns(
signal=pl.when(pl.col("p_long").is_not_null() & (pl.col(score_col) > pl.col("p_long")))
.then(pl.lit(1).cast(pl.Int8))
.otherwise(pl.lit(0).cast(pl.Int8))
)
drop_cols = ["_date", "_raw_p_long", "_is_first", "_p_long_seed", "p_long"]
if stay_q is not None:
df = df.with_columns(
_p_stay_seed=pl.when(pl.col("_is_first")).then(pl.col("_raw_p_stay")).otherwise(None),
).with_columns(
stay_thresh=pl.col("_p_stay_seed").forward_fill().over([asset_col, "_date"]),
)
drop_cols = [*drop_cols, "_raw_p_stay", "_p_stay_seed"]
return df.drop(drop_cols)
def _decile_long_short(
predictions: pl.DataFrame,
n_quantiles: int = 10,
score_col: str = "y_score",
time_col: str = "timestamp",
asset_col: str = "symbol",
) -> pl.DataFrame:
"""Academic factor portfolio: long top decile, short bottom decile.
Args:
predictions: DataFrame with [timestamp, asset, y_score]
n_quantiles: Number of quantile bins (10=decile, 5=quintile)
score_col: Score column
time_col: Timestamp column
asset_col: Asset column
Returns:
DataFrame with [timestamp, asset, weight]
"""
df = predictions.sort(time_col)
# Cross-sectional quantile rank per timestamp
df = df.with_columns(
cs_rank=pl.col(score_col).rank(method="ordinal", descending=True).over(time_col),
n_assets=pl.col(score_col).count().over(time_col),
)
# Determine top and bottom quantile thresholds
df = df.with_columns(
top_cutoff=(pl.col("n_assets") / n_quantiles).floor().cast(pl.Int64).clip(lower_bound=1),
)
# Top quantile = long, bottom quantile = short
df = df.with_columns(
signal=pl.when(pl.col("cs_rank") <= pl.col("top_cutoff"))
.then(pl.lit(1).cast(pl.Int8))
.when(pl.col("cs_rank") > pl.col("n_assets") - pl.col("top_cutoff"))
.then(pl.lit(-1).cast(pl.Int8))
.otherwise(pl.lit(0).cast(pl.Int8))
)
return _signals_to_equal_weights(df, time_col, asset_col)
def build_target_weights_from_config(
predictions: pl.DataFrame,
config: dict,
score_col: str = "y_score",
time_col: str = "timestamp",
asset_col: str = "symbol",
) -> pl.DataFrame:
"""Config-dict dispatcher for signal conversion and weight construction.
Dispatches to existing methods based on config["method"]:
- "equal_weight_top_k": top_k assets, equal weight
- "score_weighted_top_k": top_k assets, score-proportional weight
- "cross_sectional_percentile": percentile-based selection, equal weight
- "per_symbol_rolling_percentile": per-symbol time-series rolling quantile,
daily-anchored; long_q + lookback_days + bars_per_day
- "fixed_threshold": threshold-based selection, equal weight
- "decile_long_short": top/bottom decile, equal weight (academic factor)
- "quintile_long_short": top/bottom quintile, equal weight
Config dict keys:
method (str): One of the methods above
top_k (int): For top-k methods
long_short (bool): For top-k methods
percentile (float): For cross-sectional percentile (e.g., 90.0)
threshold (float): For fixed threshold
n_quantiles (int): For decile/quintile methods (default 10)
Returns:
DataFrame with [timestamp, asset, weight]
"""
method = config["method"]
long_short = config.get("long_short", False)
direction = str(config.get("direction", "long_only")).strip().lower()
def _apply_direction(weights: pl.DataFrame) -> pl.DataFrame:
if direction == "long_only":
return weights
if direction == "short_only":
return weights.with_columns((-pl.col("weight")).alias("weight"))
msg = f"Unknown signal direction: {direction}"
raise ValueError(msg)
if method in ("equal_weight_top_k", "score_weighted_top_k", "inverse_vol"):
return _apply_direction(
build_target_weights(
predictions,
method=method,
top_k=config.get("top_k", 10),
long_short=long_short,
score_col=score_col,
time_col=time_col,
asset_col=asset_col,
)
)
elif method == "cross_sectional_percentile":
percentile = config.get("percentile", 90.0)
signal_type = "long_short" if long_short else "long_only"
df_with_signal = cross_sectional_percentile_signal(
predictions,
percentile=percentile,
score_col=score_col,
time_col=time_col,
signal_type=signal_type,
)
return _apply_direction(_signals_to_equal_weights(df_with_signal, time_col, asset_col))
elif method == "per_symbol_rolling_percentile":
long_q = float(config.get("long_q", 0.80))
lookback_days = int(config.get("lookback_days", 20))
bars_per_day = int(config.get("bars_per_day", 390))
signal_type = "long_short" if long_short else "long_only"
df_with_signal = per_symbol_rolling_percentile_signal(
predictions,
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=signal_type,
)
return _apply_direction(_signals_to_equal_weights(df_with_signal, time_col, asset_col))
elif method == "fixed_threshold":
threshold = config.get("threshold", 0.0)
signal_type = "long_short" if long_short else "long_only"
df_with_signal = fixed_threshold_signal(
predictions,
threshold=threshold,
score_col=score_col,
signal_type=signal_type,
)
return _apply_direction(_signals_to_equal_weights(df_with_signal, time_col, asset_col))
elif method in ("decile_long_short", "quintile_long_short"):
n_q = config.get("n_quantiles", 10 if method == "decile_long_short" else 5)
return _apply_direction(
_decile_long_short(
predictions,
n_quantiles=n_q,
score_col=score_col,
time_col=time_col,
asset_col=asset_col,
)
)
else:
msg = f"Unknown signal method: {method}"
raise ValueError(msg)
```Exibido na íntegra, com atribuição conforme a licença da fonte. Licença: MIT
Este resumo foi escrito pelo agente de pesquisa da Stratmill com base no original; não é uma cópia da fonte.