Yuxuan Yan
71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
"""Map signal values to discrete position actions."""
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
|
|
@dataclass
|
|
class PositionAction:
|
|
"""A target action for a single stock on a single bar."""
|
|
|
|
action: str # "buy", "sell", or "hold"
|
|
size_pct: float = 0.0 # target portfolio fraction for buys
|
|
|
|
|
|
class ThresholdBuilder:
|
|
"""Open on strong positive signal, close on strong negative signal."""
|
|
|
|
def __init__(
|
|
self,
|
|
buy_threshold: float = 0.02,
|
|
sell_threshold: float = -0.02,
|
|
size_pct: float = 0.95,
|
|
):
|
|
self.buy_threshold = buy_threshold
|
|
self.sell_threshold = sell_threshold
|
|
self.size_pct = size_pct
|
|
|
|
def build(self, signal_value: float, in_position: bool) -> PositionAction:
|
|
if not in_position and signal_value >= self.buy_threshold:
|
|
return PositionAction("buy", self.size_pct)
|
|
if in_position and signal_value <= self.sell_threshold:
|
|
return PositionAction("sell", 0.0)
|
|
return PositionAction("hold", 0.0)
|
|
|
|
|
|
class RankEqualWeightBuilder:
|
|
"""Rank all stocks by signal. Buy top N at equal weight. Sell if drops out.
|
|
|
|
Called once per bar with ALL stock signals. Returns per-stock actions.
|
|
"""
|
|
|
|
def __init__(self, top_n: int = 5, min_signal: Optional[float] = None):
|
|
self.top_n = top_n
|
|
self.min_signal = min_signal # optional floor — skip stocks with signal below this
|
|
|
|
def build(self, signals: dict[str, float]) -> dict[str, PositionAction]:
|
|
"""Rank stocks by signal (descending). Top N get 'buy', rest get 'sell' if held.
|
|
|
|
Args:
|
|
signals: {symbol: signal_value} for all stocks on this bar.
|
|
|
|
Returns:
|
|
{symbol: PositionAction}
|
|
"""
|
|
# Filter by min_signal if set
|
|
if self.min_signal is not None:
|
|
signals = {s: v for s, v in signals.items() if v >= self.min_signal}
|
|
|
|
# Sort by signal descending
|
|
ranked = sorted(signals.items(), key=lambda x: x[1], reverse=True)
|
|
|
|
top_symbols = set(sym for sym, _ in ranked[:self.top_n])
|
|
size_pct = 1.0 / self.top_n if self.top_n > 0 else 0.0
|
|
|
|
actions = {}
|
|
for sym in signals:
|
|
if sym in top_symbols:
|
|
actions[sym] = PositionAction("buy", size_pct)
|
|
else:
|
|
actions[sym] = PositionAction("sell", 0.0)
|
|
return actions
|