169 lines
6.8 KiB
Python
169 lines
6.8 KiB
Python
"""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) |