Yuxuan Yan
70 lines
2.3 KiB
Python
70 lines
2.3 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: Optional[int] = None, top_pct: float = 0.2, min_signal: Optional[float] = None):
|
|
self.top_n = top_n
|
|
self.top_pct = top_pct
|
|
self.min_signal = min_signal
|
|
|
|
def build(self, signals: dict[str, float]) -> dict[str, 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)
|
|
|
|
# Determine top N: explicit count or percentage of available stocks
|
|
if self.top_n is not None:
|
|
n = self.top_n
|
|
else:
|
|
n = max(1, int(len(signals) * self.top_pct))
|
|
|
|
top_symbols = set(sym for sym, _ in ranked[:n])
|
|
size_pct = 1.0 / n if 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
|