"""Market-neutral top/bottom long-short strategy for cross-sectional signals. Captures the cross-sectional long-short spread net of costs: buys the top-ranked ``topk`` names and shorts the bottom-ranked ``topk`` names, equal-weight per side, sized to ``risk_degree`` of total value per side. Rebalances daily to the current rank (dropout-free: the book converges to the latest top/bottom sets). The long and short legs use equal notional per side (gross exposure ~2x ``risk_degree`` of NAV, i.e. approximately market neutral before transaction costs). Benchmark neutrality (SPY beta ~ 0) is the secondary sanity metric. """ from __future__ import annotations from typing import List import copy 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__ = ["TopBottomDropoutStrategy"] DEFAULT_SHORT_LEG = True DEFAULT_REBALANCE_DAILY = True class TopBottomDropoutStrategy(BaseSignalStrategy): """Long top-k / short bottom-k equal-weight market-neutral book. Parameters ---------- topk : number of names on each side (long top-k and short bottom-k). short_leg : whether to open the short side (if False, long-only topk). rebalance_daily : if True rebalance to current rank every day; else keep positions and only refresh on score changes (dropout-style). risk_degree : fraction of total value deployed per side. """ def __init__( self, *, topk: int = 10, short_leg: bool = DEFAULT_SHORT_LEG, rebalance_daily: bool = DEFAULT_REBALANCE_DAILY, **kwargs, ): super().__init__(**kwargs) self.topk = topk self.short_leg = short_leg self.rebalance_daily = rebalance_daily self._prev_longs = set() self._prev_shorts = set() def generate_trade_decision(self, execute_result=None): trade_step = self.trade_calendar.get_trade_step() trade_start_time, trade_end_time = self.trade_calendar.get_step_time(trade_step) pred_start_time, pred_end_time = self.trade_calendar.get_step_time(trade_step, shift=1) pred_score = self.signal.get_signal(start_time=pred_start_time, end_time=pred_end_time) 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) # rank all names; topk longs and topk shorts ranked = pred_score.sort_values(ascending=False) longs = list(ranked.index[: self.topk]) shorts = list(ranked.index[-self.topk :]) if self.short_leg else [] current_temp: "object" = copy.deepcopy(self.trade_position) current_codes = set(current_temp.get_stock_list()) holdings = {c: current_temp for c in current_codes if abs(current_temp.get_stock_amount(c)) > 1e-6} sell_orders: List[Order] = [] buy_orders: List[Order] = [] def _tradable(code, direction): try: return self.trade_exchange.is_stock_tradable( stock_id=code, start_time=trade_start_time, end_time=trade_end_time, direction=direction ) except TypeError: return self.trade_exchange.is_stock_tradable( stock_id=code, start_time=trade_start_time, end_time=trade_end_time ) # determine target set (long/short) target_longs = set(longs) target_shorts = set(shorts) # close positions not in the target book for code in list(holdings): if code in target_longs or code in target_shorts: continue amt = abs(current_temp.get_stock_amount(code)) o = Order( stock_id=code, amount=amt, start_time=trade_start_time, end_time=trade_end_time, direction=Order.SELL if code in target_longs else Order.SELL, ) if self.trade_exchange.check_order(o): sell_orders.append(o) self.trade_exchange.deal_order(o, position=current_temp) # equal-weight notional per side total_value = current_temp.get_cash() for code, pos in holdings.items(): if code in target_longs or code in target_shorts: mark = self.trade_exchange.get_deal_price( stock_id=code, start_time=trade_start_time, end_time=trade_end_time, direction=Order.SELL ) if mark is not None and mark == mark: total_value += abs(current_temp.get_stock_amount(code)) * mark side_notional = total_value * self.risk_degree / max(1, self.topk) for code in longs: if code in holdings and abs(current_temp.get_stock_amount(code)) > 1e-6: continue px = self.trade_exchange.get_deal_price( stock_id=code, start_time=trade_start_time, end_time=trade_end_time, direction=Order.BUY ) if px is None or px != px or px <= 0: continue amount = side_notional / px factor = self.trade_exchange.get_factor( stock_id=code, start_time=trade_start_time, end_time=trade_end_time ) amount = self.trade_exchange.round_amount_by_trade_unit(amount, factor) o = Order( stock_id=code, amount=amount, start_time=trade_start_time, end_time=trade_end_time, direction=Order.BUY, ) if self.trade_exchange.check_order(o): buy_orders.append(o) if self.short_leg: for code in shorts: if code in holdings and abs(current_temp.get_stock_amount(code)) > 1e-6: continue px = self.trade_exchange.get_deal_price( stock_id=code, start_time=trade_start_time, end_time=trade_end_time, direction=Order.SELL ) if px is None or px != px or px <= 0: continue amount = side_notional / px factor = self.trade_exchange.get_factor( stock_id=code, start_time=trade_start_time, end_time=trade_end_time ) amount = self.trade_exchange.round_amount_by_trade_unit(amount, factor) o = Order( stock_id=code, amount=amount, start_time=trade_start_time, end_time=trade_end_time, direction=Order.SELL, ) if self.trade_exchange.check_order(o): sell_orders.append(o) return TradeDecisionWO(sell_orders + buy_orders, self)