218 lines
8.9 KiB
Python
218 lines
8.9 KiB
Python
"""Optimal-stopping / stochastic-control strategy for cross-sectional signals.
|
|
|
|
Entry is a control policy: a symbol opens a position only when its cross-sectional
|
|
signal percentile is at or above ``entry_pct`` (i.e. it is one of the top-ranked
|
|
names) and the portfolio has fewer than ``topk`` open positions.
|
|
|
|
Exit is an optimal-stopping rule: a held position is stopped (closed) when its
|
|
signal percentile falls below ``exit_pct`` (the continuation value of holding is
|
|
no longer worth the risk), OR after ``max_hold_days`` (time stop / finite
|
|
horizon), OR when the position P&L breaches ``sl`` (loss control) and the
|
|
position has been held at least ``min_hold_days``.
|
|
|
|
Sizing is fixed ``notional`` per position (equal-weight control), unlike the
|
|
TopkDropout cash-allocation heuristic.
|
|
|
|
Wired into qrun workflows like any ``BaseStrategy`` (see ``PortAnaRecord``
|
|
config). Mirrors the API usage of qlib's ``TopkDropoutStrategy``: ``Order``/
|
|
``OrderDir`` from ``qlib.backtest.decision``, ``trade_calendar`` /
|
|
``trade_exchange`` / ``trade_position`` injected by the backtest executor.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import List
|
|
|
|
import pandas as pd
|
|
|
|
from qlib.backtest import Order
|
|
from qlib.backtest.decision import OrderDir, TradeDecisionWO
|
|
from qlib.contrib.strategy.signal_strategy import BaseSignalStrategy
|
|
|
|
__all__ = ["OptimalStopControl"]
|
|
|
|
DEFAULT_NOTIONAL = 20_000.0
|
|
DEFAULT_ENTRY_PCT = 0.80
|
|
DEFAULT_EXIT_PCT = 0.50
|
|
DEFAULT_MAX_HOLD_DAYS = 10
|
|
DEFAULT_MIN_HOLD_DAYS = 2
|
|
DEFAULT_SL = -0.06
|
|
|
|
|
|
class OptimalStopControl(BaseSignalStrategy):
|
|
"""Optimal-stopping long-only strategy over a cross-sectional signal.
|
|
|
|
Parameters
|
|
----------
|
|
topk : max number of concurrent positions.
|
|
entry_pct : min cross-sectional score percentile required to OPEN (0..1).
|
|
exit_pct : held positions are stopped when score percentile < exit_pct.
|
|
max_hold_days : hard time stop (finite-horizon close).
|
|
min_hold_days : minimum holding days before stop-loss is evaluated.
|
|
notional : $ per position (equal-weight control).
|
|
sl : stop-loss threshold as fraction of entry price (<= 0), disabled if 0.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
signal=None,
|
|
topk: int = 10,
|
|
entry_pct: float = DEFAULT_ENTRY_PCT,
|
|
exit_pct: float = DEFAULT_EXIT_PCT,
|
|
max_hold_days: int = DEFAULT_MAX_HOLD_DAYS,
|
|
min_hold_days: int = DEFAULT_MIN_HOLD_DAYS,
|
|
notional: float = DEFAULT_NOTIONAL,
|
|
sl: float = DEFAULT_SL,
|
|
risk_degree: float = 0.95,
|
|
trade_exchange=None,
|
|
level_infra=None,
|
|
common_infra=None,
|
|
**kwargs,
|
|
):
|
|
super().__init__(
|
|
signal=signal,
|
|
trade_exchange=trade_exchange,
|
|
level_infra=level_infra,
|
|
common_infra=common_infra,
|
|
**kwargs,
|
|
)
|
|
self.topk = topk
|
|
self.entry_pct = entry_pct
|
|
self.exit_pct = exit_pct
|
|
self.max_hold_days = max_hold_days
|
|
self.min_hold_days = min_hold_days
|
|
self.notional = notional
|
|
self.sl = sl
|
|
|
|
# ------------------------------------------------------------------ utils
|
|
@staticmethod
|
|
def _pct_rank(score: pd.Series) -> pd.Series:
|
|
return score.rank(pct=True)
|
|
|
|
def _entry_price(self, pos) -> float:
|
|
# Position stores avg entry price under key "price" (see Position.position)
|
|
price = pos.position.get("price")
|
|
if price is None:
|
|
price = pos.get_stock_amount("price")
|
|
return float(price)
|
|
|
|
def _pnl_pct(self, pos, mark: float) -> float:
|
|
entry = self._entry_price(pos)
|
|
if not entry or entry != entry:
|
|
return 0.0
|
|
return mark / entry - 1.0
|
|
|
|
def _is_tradable(self, code, start, end, direction) -> bool:
|
|
try:
|
|
return self.trade_exchange.is_stock_tradable(
|
|
stock_id=code, start_time=start, end_time=end, direction=direction
|
|
)
|
|
except TypeError: # some exchanges take no direction kwarg
|
|
return self.trade_exchange.is_stock_tradable(stock_id=code, start_time=start, end_time=end)
|
|
|
|
# ------------------------------------------------------------ decision
|
|
def generate_trade_decision(self, execute_result=None):
|
|
trade_step = self.trade_calendar.get_trade_step()
|
|
trade_start, trade_end = self.trade_calendar.get_step_time(trade_step)
|
|
pred_start, pred_end = self.trade_calendar.get_step_time(trade_step, shift=1)
|
|
pred_score = self.signal.get_signal(start_time=pred_start, end_time=pred_end)
|
|
if isinstance(pred_score, pd.DataFrame):
|
|
pred_score = pred_score.iloc[:, 0]
|
|
if pred_score is None or len(pred_score) == 0:
|
|
return TradeDecisionWO([], self)
|
|
|
|
pct = self._pct_rank(pred_score)
|
|
time_per_step = self.trade_calendar.get_freq()
|
|
current_temp = __import__("copy").deepcopy(self.trade_position)
|
|
|
|
holdings = {}
|
|
for code in current_temp.get_stock_list():
|
|
if abs(current_temp.get_stock_amount(code)) > 1e-6:
|
|
holdings[code] = current_temp
|
|
|
|
# ---- optimal stopping: close held positions -----------------------
|
|
sell_orders: List[Order] = []
|
|
closed_today = set()
|
|
kept = {}
|
|
for code, pos in holdings.items():
|
|
held = current_temp.get_stock_count(code, bar=time_per_step)
|
|
mark = self.trade_exchange.get_deal_price(
|
|
stock_id=code, start_time=trade_start, end_time=trade_end, direction=Order.SELL
|
|
)
|
|
if mark is None or mark != mark:
|
|
continue
|
|
rank = pct.get(code, 0.0)
|
|
stop_pnl = held >= self.min_hold_days and self.sl < 0 and self._pnl_pct(pos, mark) <= self.sl
|
|
if held >= self.max_hold_days or rank < self.exit_pct or stop_pnl:
|
|
amt = abs(current_temp.get_stock_amount(code))
|
|
o = Order(stock_id=code, amount=amt, start_time=trade_start,
|
|
end_time=trade_end, direction=Order.SELL)
|
|
if self.trade_exchange.check_order(o):
|
|
sell_orders.append(o)
|
|
self.trade_exchange.deal_order(o, position=current_temp)
|
|
closed_today.add(code)
|
|
else:
|
|
kept[code] = mark
|
|
|
|
# ---- equal-weight control: target notional per name -----------------
|
|
# candidate opens: top-ranked names whose signal pct >= entry_pct
|
|
rank_desc = pred_score.sort_values(ascending=False)
|
|
held_codes = set(kept)
|
|
opens = []
|
|
for sym in rank_desc.index:
|
|
if len(opens) >= self.topk:
|
|
break
|
|
if sym in held_codes:
|
|
continue
|
|
if pct.get(sym, 0.0) < self.entry_pct:
|
|
continue
|
|
if not self._is_tradable(sym, trade_start, trade_end, OrderDir.BUY):
|
|
continue
|
|
opens.append(sym)
|
|
|
|
targets = held_codes | set(opens)
|
|
if not targets:
|
|
return TradeDecisionWO(sell_orders, self)
|
|
|
|
# total value (cash + marked positions) -> per-target notional
|
|
total_value = current_temp.get_cash()
|
|
for code, mark in kept.items():
|
|
total_value += abs(current_temp.get_stock_amount(code)) * mark
|
|
|
|
target_notional = total_value * self.risk_degree / max(1, len(targets))
|
|
|
|
# ---- rebalance kept positions toward target weight ------------------
|
|
buy_orders: List[Order] = []
|
|
for code, mark in kept.items():
|
|
cur = abs(current_temp.get_stock_amount(code)) * mark
|
|
diff_notional = target_notional - cur
|
|
if abs(diff_notional) / target_notional < 0.02:
|
|
continue # skip tiny rebalances
|
|
amount_delta = diff_notional / mark
|
|
direction = Order.BUY if amount_delta > 0 else Order.SELL
|
|
o = Order(stock_id=code, amount=abs(amount_delta), start_time=trade_start,
|
|
end_time=trade_end, direction=direction)
|
|
if self.trade_exchange.check_order(o):
|
|
(buy_orders if direction == Order.BUY else sell_orders).append(o)
|
|
self.trade_exchange.deal_order(o, position=current_temp)
|
|
|
|
# ---- open new positions at target weight ----------------------------
|
|
for sym in opens:
|
|
px = self.trade_exchange.get_deal_price(
|
|
stock_id=sym, start_time=trade_start, end_time=trade_end, direction=OrderDir.BUY
|
|
)
|
|
if px is None or px != px or px <= 0:
|
|
continue
|
|
amount = target_notional / px
|
|
factor = self.trade_exchange.get_factor(
|
|
stock_id=sym, start_time=trade_start, end_time=trade_end
|
|
)
|
|
amount = self.trade_exchange.round_amount_by_trade_unit(amount, factor)
|
|
o = Order(stock_id=sym, amount=amount, start_time=trade_start,
|
|
end_time=trade_end, direction=Order.BUY)
|
|
if self.trade_exchange.check_order(o):
|
|
buy_orders.append(o)
|
|
|
|
return TradeDecisionWO(sell_orders + buy_orders, self)
|