wehub-resource-sync
615 lines
22 KiB
Python
615 lines
22 KiB
Python
"""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
|
||
- Inverse volatility: equal-weight placeholder (full impl in Ch17)
|
||
"""
|
||
|
||
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),
|
||
)
|
||
|
||
# Effective top_k (can't select more assets than available)
|
||
df = df.with_columns(
|
||
eff_k=pl.min_horizontal(pl.lit(top_k), pl.col("n_assets")),
|
||
)
|
||
|
||
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)
|