78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
"""Minimal OptimalStopControl strategy — a stub of tac_qlib/contrib/strategy/optimal_stop.py.
|
|
|
|
Subclasses qlib's BaseSignalStrategy; override `generate_trade_decision` to build
|
|
`qlib.backtest.Order`s and return a `TradeDecisionWO`. The real implementation
|
|
gates entry by cross-sectional signal percentile, exits by percentile / time /
|
|
stop-loss, and sizes equal-weight with `risk_degree` control.
|
|
|
|
Wire into a workflow YAML under PortAnaRecord.config.strategy:
|
|
|
|
strategy:
|
|
class: OptimalStopControl
|
|
module_path: tac_qlib.contrib.strategy.optimal_stop
|
|
kwargs:
|
|
signal: "<PRED>"
|
|
topk: 10
|
|
entry_pct: 0.85
|
|
exit_pct: 0.7
|
|
max_hold_days: 10
|
|
min_hold_days: 2
|
|
sl: -0.08
|
|
risk_degree: 0.95
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import numpy as np
|
|
|
|
from qlib.backtest import Order
|
|
from qlib.backtest.decision import OrderDir, TradeDecisionWO
|
|
from qlib.contrib.strategy.signal_strategy import BaseSignalStrategy
|
|
|
|
|
|
class OptimalStopControl(BaseSignalStrategy):
|
|
def __init__(
|
|
self,
|
|
*,
|
|
topk: int = 10,
|
|
entry_pct: float = 0.85,
|
|
exit_pct: float = 0.7,
|
|
max_hold_days: int = 10,
|
|
min_hold_days: int = 2,
|
|
sl: float = -0.08,
|
|
risk_degree: float = 0.95,
|
|
**kwargs: Any,
|
|
):
|
|
super().__init__(**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.sl = sl
|
|
self.risk_degree = risk_degree
|
|
|
|
def generate_trade_decision(self, execute_result=None):
|
|
"""Build orders for one trade step (minimal sketch — see repo impl)."""
|
|
trade_step = self.trade_calendar.get_trade_step()
|
|
# signal is known at t-1 via shift=-1 in the signal object
|
|
start_time, end_time = self.trade_calendar.get_step_time(trade_step)
|
|
pred_start, pred_end = self.trade_calendar.get_step_time(trade_step - 1)
|
|
pred = self.signal.get_signal(start_time=pred_start, end_time=pred_end)
|
|
|
|
orders: List[Order] = []
|
|
if pred is not None and len(pred):
|
|
# take the top-k by cross-sectional percentile, equal-weight size
|
|
cross = pred.groupby(level=0).rank(pct=True) # 0..1 per day
|
|
keep = pred.index[cross >= 1.0 - self.entry_pct]
|
|
for inst, (dt, _instr) in zip(keep, keep):
|
|
price = self.trade_exchange.get_close(inst, end_time) or 1.0
|
|
qty = int((self.risk_degree * self.trade_exchange.account.cash) / (self.topk * price))
|
|
if qty > 0:
|
|
orders.append(
|
|
Order(inst, qty, start_time, end_time, direction=OrderDir.BUY, type="market")
|
|
)
|
|
return TradeDecisionWO(orders, self)
|