zhaoli/tac-exp-dev
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

start experiment 73 (exp/73-scheduled-algo-trade-run-for-target-date)

Browse Source
This commit is contained in:
zhaoli
2026-09-14 12:10:05 +00:00
parent fd5382caa4
commit 73b66a00f6
21 changed files with 3030 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code/tac-qlib/tac_qlib/contrib/backtest/tradeac_exchange.py
+432
View File
@@ -0,0 +1,432 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
"""
TradeACExchange
A short/borrow enabled Exchange implementation built on top of qlib.backtest.exchange.Exchange.
This exchange adds simple, configurable margin logic (initial/maintenance), borrowing support for
shorts, a borrow fee, and a lightweight SMA (Special Memorandum Account) concept to emulate
behaviors similar to brokers such as IBKR and Alpaca for backtesting purposes.
Notes / limitations
- This implementation is intentionally lightweight and conservative: it implements the
key behaviors needed for strategy/backtest experiments (allowing short selling, computing
margin requirements, performing margin-call checks, and tracking SMA-like excess equity).
- It makes some simplifying assumptions compared to real brokers (no per-product house margins,
simplified SMA bookkeeping, borrow availability modeled only by a per-symbol boolean/limit).
- The Position class in qlib.backtest.position was not changed. To support shorts we update the
position.position dict directly when necessary. This keeps integration simple but bypasses some
internal Position helpers. Use with care.
API additions
- allow_short: enable short selling (bool)
- initial_margin_long/short: fraction required to open a position
- maintenance_margin_long/short: fraction required to keep a position
- borrow_fee_rate: periodic borrow fee applied on short value (applied at trade time as additional cost)
- borrowable: dict mapping stock_id -> bool or float (max borrowable shares). Symbols missing from
the dict follow `borrow_default` (default True = unlimited; set False for a strict whitelist)
- get_sma(position): returns SMA-like excess equity available as "buying power credit"
- check_margin_call(position): returns True if position is below maintenance requirement
"""
from __future__ import annotations
from typing import Any, Dict, Optional, Tuple
import numpy as np
from qlib.backtest.decision import Order
from qlib.backtest.exchange import Exchange
from qlib.backtest.position import BasePosition
class TradeACExchange(Exchange):
"""An exchange that supports short selling / borrowing and basic margin rules.
The implementation aims to be compatible with the Exchange API used by Account and
Position classes in qlib.backtest. It overrides only the minimum methods required to
enable short/borrow behavior and margin calculations.
"""
def __init__(
self,
*args: Any,
allow_short: bool = True,
initial_margin_long: float = 0.5,
initial_margin_short: float = 0.5,
maintenance_margin_long: float = 0.25,
maintenance_margin_short: float = 0.3,
borrow_fee_rate: float = 0.0,
borrowable: Optional[Dict[str, float]] = None,
borrow_default: bool = True,
sma_enabled: bool = True,
**kwargs: Any,
) -> None:
"""Create TradeACExchange.
Parameters mirror Exchange with additional tradeac-specific options.
"""
super().__init__(*args, **kwargs)
self.allow_short = allow_short
self.initial_margin_long = initial_margin_long
self.initial_margin_short = initial_margin_short
self.maintenance_margin_long = maintenance_margin_long
self.maintenance_margin_short = maintenance_margin_short
self.borrow_fee_rate = borrow_fee_rate
# borrowable can be a dict with per-symbol max borrowable amount, or None (unlimited)
self.borrowable = borrowable or {}
# borrow_default: policy for symbols absent from `borrowable`.
# True -> unlisted symbols are unlimited-borrowable (legacy behavior)
# False -> unlisted symbols are NOT borrowable; only listed ones can be shorted
self.borrow_default = bool(borrow_default)
# sma_enabled: whether to expose lightweight SMA calculation
self.sma_enabled = sma_enabled
# --------------------------- Helper calculations ---------------------------
def _initial_margin_requirement(self, position: BasePosition) -> float:
"""Compute the initial margin requirement (money) for the given position.
We treat longs and shorts separately and sum their required initial margins.
"""
im_req = 0.0
for sid in position.get_stock_list():
amt = position.get_stock_amount(sid)
price = position.get_stock_price(sid)
val = amt * price
if val > 0:
im_req += abs(val) * self.initial_margin_long
elif val < 0:
im_req += abs(val) * self.initial_margin_short
return im_req
def _maintenance_margin_requirement(self, position: BasePosition) -> float:
"""Compute the maintenance margin requirement (money) for the given position."""
mm_req = 0.0
for sid in position.get_stock_list():
amt = position.get_stock_amount(sid)
price = position.get_stock_price(sid)
val = amt * price
if val > 0:
mm_req += abs(val) * self.maintenance_margin_long
elif val < 0:
mm_req += abs(val) * self.maintenance_margin_short
return mm_req
def get_equity(self, position: BasePosition) -> float:
"""Return account equity (position value + cash)."""
return position.calculate_value()
def get_sma(self, position: BasePosition) -> float:
"""Return a simplified SMA: excess equity above initial margin requirement.
Note: This is a synthetic/Simplified SMA used for strategy/backtest logic. Real-broker
SMA accounting (e.g. credits/debits across days) can be more complex.
"""
if not self.sma_enabled:
return 0.0
equity = self.get_equity(position)
im_req = self._initial_margin_requirement(position)
return max(0.0, equity - im_req)
def check_margin_call(self, position: BasePosition) -> bool:
"""Return True when the account is under maintenance margin (margin call).
Margin call condition here is simple: equity < maintenance requirement.
"""
equity = self.get_equity(position)
mm_req = self._maintenance_margin_requirement(position)
return equity < mm_req
def get_buying_power(self, position: BasePosition) -> float:
"""Estimate buying power for new long positions assuming opening margin requirement.
Simplified: the maximum notional long value = equity / initial_margin_long.
"""
equity = self.get_equity(position)
if self.initial_margin_long <= 0:
return 0.0
return equity / self.initial_margin_long
# --------------------------- Order / execution overrides ---------------------------
def _borrow_headroom(self, stock_id: str, current_short: float) -> float:
"""Remaining borrowable shares for `stock_id` given an already-open short of `current_short` shares.
borrowable values: bool (True=unlimited, False=not borrowable) or numeric max shares.
Missing symbols follow `borrow_default` (True = unlimited when allow_short is enabled).
"""
if not self.allow_short:
return 0.0
v = self.borrowable.get(stock_id, self.borrow_default)
if isinstance(v, bool):
return float("inf") if v else 0.0
try:
limit = float(v)
except (TypeError, ValueError):
return float("inf")
return max(0.0, limit - max(current_short, 0.0))
def _calc_trade_info_by_order(
self,
order: Order,
position: Optional[BasePosition],
dealt_order_amount: Dict[str, float],
) -> Tuple[float, float, float]:
"""Override to allow (optionally) short selling and to apply borrow fees.
The original Exchange implementation forbids selling more than you own. Here we allow
sell orders to create/expand short positions when allow_short is True. We still rely on
most base logic (price discovery, impact, cost calculation) by calling super(), but we
adjust the sell-side clipping behavior before delegating to the base implementation.
"""
# When selling and shorts are allowed, temporarily relax the clipping logic in the base
# implementation by monkey-patching current position check. Simpler: replicate minimal
# parts of logic from Exchange._calc_trade_info_by_order with the key change.
# Get basic trade price & volume info using Exchange helpers
trade_price = float(self.get_deal_price(order.stock_id, order.start_time, order.end_time, direction=order.direction))
total_trade_val = float(self.get_volume(order.stock_id, order.start_time, order.end_time) or 0.0) * trade_price
order.factor = self.get_factor(order.stock_id, order.start_time, order.end_time)
order.deal_amount = order.amount # attempt full
# volume clipping (same as base)
self._clip_amount_by_volume(order, dealt_order_amount)
# approximate adjusted cost ratio based on liquidity
if not total_trade_val or np.isnan(total_trade_val) or total_trade_val <= 0:
adj_cost_ratio = self.impact_cost
else:
trade_val_tmp = order.deal_amount * trade_price
adj_cost_ratio = self.impact_cost * (trade_val_tmp / total_trade_val) ** 2
# Differentiate buy / sell
if order.direction == Order.SELL:
cost_ratio = self.close_cost + adj_cost_ratio
current_amount = (
position.get_stock_amount(order.stock_id) if (position is not None and position.check_stock(order.stock_id)) else 0.0
)
long_held = max(current_amount, 0.0)
short_open = max(-current_amount, 0.0)
if position is not None:
if not self.allow_short:
# clip by current holdings only
if not np.isclose(order.deal_amount, current_amount):
order.deal_amount = self.round_amount_by_trade_unit(
min(long_held, order.deal_amount), order.factor
)
else:
# allow selling beyond holdings up to the remaining borrow limit;
# later when updating the position we create/expand a short if necessary.
max_sell = long_held + self._borrow_headroom(order.stock_id, short_open)
if order.deal_amount > max_sell and not np.isclose(order.deal_amount, max_sell):
order.deal_amount = self.round_amount_by_trade_unit(max_sell, order.factor)
elif order.direction == Order.BUY:
cost_ratio = self.open_cost + adj_cost_ratio
if position is not None:
cash = position.get_cash()
trade_val = order.deal_amount * trade_price
if cash < max(trade_val * cost_ratio, self.min_cost):
order.deal_amount = 0
self.logger.debug(f"Order clipped due to cost higher than cash: {order}")
elif cash < trade_val + max(trade_val * cost_ratio, self.min_cost):
max_buy_amount = self._get_buy_amount_by_cash_limit(trade_price, cash, cost_ratio)
order.deal_amount = self.round_amount_by_trade_unit(min(max_buy_amount, order.deal_amount), order.factor)
self.logger.debug(f"Order clipped due to cash limitation: {order}")
else:
order.deal_amount = self.round_amount_by_trade_unit(order.deal_amount, order.factor)
else:
order.deal_amount = self.round_amount_by_trade_unit(order.deal_amount, order.factor)
else:
raise NotImplementedError("order direction {} error".format(order.direction))
# compute final trade_val & trade_cost
trade_val = order.deal_amount * trade_price
# base trade_cost
trade_cost = max(trade_val * cost_ratio, self.min_cost)
# apply borrow fee only on the net-new short portion of the sell
if order.direction == Order.SELL and self.allow_short:
new_short = max(0.0, order.deal_amount - long_held)
trade_cost += new_short * trade_price * self.borrow_fee_rate
if trade_val <= 1e-5:
trade_cost = 0
return trade_price, trade_val, trade_cost
def deal_order(
self,
order: Order,
trade_account: Optional[Any] = None,
position: Optional[BasePosition] = None,
dealt_order_amount: Dict[str, float] = None,
) -> Tuple[float, float, float]:
"""Deal order and handle short position bookkeeping.
This method mirrors Exchange.deal_order but when a position is provided and shorts are
allowed it will update the Position.position dict directly to support negative amounts.
"""
if dealt_order_amount is None:
dealt_order_amount = {}
if not self.check_order(order):
order.deal_amount = 0.0
self.logger.debug(f"Order failed due to trading limitation: {order}")
return 0.0, 0.0, np.nan
if trade_account is not None and position is not None:
raise ValueError("trade_account and position can only choose one")
pos = position or (trade_account.current_position if trade_account is not None else None)
trade_price, trade_val, trade_cost = self._calc_trade_info_by_order(order, pos, dealt_order_amount)
if trade_val > 1e-5:
if trade_account is not None:
cp = trade_account.current_position
if not cp.skip_update():
held = cp.check_stock(order.stock_id)
# Account-level bookkeeping (turnover/cost/returns). Mirrors
# Account._update_state_from_order except for fresh short sales,
# where no prior price exists to compute order profit from.
if order.direction == Order.SELL and not held:
trade_account.accum_info.add_turnover(trade_val)
trade_account.accum_info.add_cost(trade_cost)
trade_account.accum_info.add_return_value(0.0)
if order.direction == Order.SELL:
# sell: update account state first (stock entry may be deleted)
if held:
trade_account._update_state_from_order(order, trade_val, trade_cost, trade_price)
self._position_sell(cp, order, trade_val, trade_cost, trade_price)
else:
# buy: update position first (entry may be created), then account state
# A buy that covers a short to exactly flat deletes the entry inside
# _position_buy; re-seed a transient zero-amount stub so the
# account's order-profit lookup still finds the trade price,
# then drop it (_update_state_from_order never mutates entries).
sid = order.stock_id
had_entry = isinstance(cp.position.get(sid), dict)
self._position_buy(cp, order, trade_val, trade_cost, trade_price)
covered_to_flat = had_entry and not isinstance(cp.position.get(sid), dict)
if covered_to_flat:
cp.position[sid] = {"amount": 0.0, "price": trade_price, "weight": 0}
trade_account._update_state_from_order(order, trade_val, trade_cost, trade_price)
if covered_to_flat:
cp.position.pop(sid, None)
elif position is not None:
if order.direction == Order.BUY:
self._position_buy(position, order, trade_val, trade_cost, trade_price)
else:
self._position_sell(position, order, trade_val, trade_cost, trade_price)
return trade_val, trade_cost, trade_price
# --------------------------- Position mutation helpers ---------------------------
def _position_buy(self, position: BasePosition, order: Order, trade_val: float, cost: float, trade_price: float) -> None:
"""Handle buy order bookkeeping against a BasePosition while supporting shorts.
Rules implemented (simplified):
- If there is an existing short (amount < 0), the buy will first cover the short.
- If covering closes the short completely, the remaining buy becomes a long
- Cash updates mimic Position._buy_stock/_sell_stock (cash decreases by trade_val+cost for buys)
"""
trade_amount = trade_val / trade_price
sid = order.stock_id
current_amount = position.get_stock_amount(sid) if position.check_stock(sid) else 0.0
# covering existing short
if current_amount < -1e-12:
# amount is negative -> we are short. Buying reduces the short.
new_amount = current_amount + trade_amount
if abs(new_amount) <= 1e-8:
# short fully covered exactly -> remove entry
if sid in position.position:
del position.position[sid]
elif new_amount > 0:
# short fully covered with leftover buy amount -> leftover becomes a long position
position.position[sid] = {"amount": new_amount, "price": trade_price, "weight": 0}
else:
# partially cover
position.position[sid]["amount"] = new_amount
position.position[sid]["price"] = trade_price
else:
# normal or increasing long
if sid not in position.position or not isinstance(position.position[sid], dict):
# initialize stock
position.position[sid] = {"amount": trade_amount, "price": trade_price, "weight": 0}
else:
position.position[sid]["amount"] = position.position[sid].get("amount", 0.0) + trade_amount
position.position[sid]["price"] = trade_price
# cash effect same as Position._buy_stock
position.position["cash"] -= trade_val + cost
def _position_sell(self, position: BasePosition, order: Order, trade_val: float, cost: float, trade_price: float) -> None:
"""Handle sell order bookkeeping against a BasePosition while supporting shorts.
Rules implemented (simplified):
- If holding enough long shares, sell will reduce/close the long position normally.
- If not holding enough long shares and shorts are allowed, the remaining sold amount will create/expand a short position.
- Cash update for sells follows Position._sell_stock logic (cash increases by trade_val - cost)
"""
trade_amount = trade_val / trade_price
sid = order.stock_id
current_amount = position.get_stock_amount(sid) if position.check_stock(sid) else 0.0
if current_amount > 1e-12:
# we have long shares; sell from them first
if trade_amount >= current_amount - 1e-8:
# selling all or more than holdings
# remove long position
if sid in position.position:
del position.position[sid]
# remaining sold amount becomes short if allowed
remain = trade_amount - current_amount
if remain > 1e-8:
if not self.allow_short:
# should not happen due to clipping earlier, but guard anyway
raise ValueError(f"Attempt to short {sid} while shorting disabled")
# create short entry
position.position[sid] = {"amount": -remain, "price": trade_price, "weight": 0}
else:
# partial sell
position.position[sid]["amount"] = current_amount - trade_amount
position.position[sid]["price"] = trade_price
else:
# currently flat or already short
if not self.allow_short:
raise ValueError(f"Attempt to short {sid} while shorting disabled")
# expand short
new_amount = current_amount - trade_amount
if sid not in position.position or not isinstance(position.position[sid], dict):
position.position[sid] = {"amount": new_amount, "price": trade_price, "weight": 0}
else:
position.position[sid]["amount"] = new_amount
position.position[sid]["price"] = trade_price
# cash effect same as Position._sell_stock
new_cash = trade_val - cost
if getattr(position, "_settle_type", None) == position.ST_CASH:
position.position["cash_delay"] = position.position.get("cash_delay", 0.0) + new_cash
else:
position.position["cash"] = position.position.get("cash", 0.0) + new_cash
# --------------------------- Borrow availability helpers ---------------------------
def is_borrowable(self, stock_id: str, amount: float) -> bool:
"""Check whether the requested amount is borrowable for the given stock.
If a borrowable dict is provided, it may contain either booleans or numeric limits (maximum borrowable shares).
Symbols absent from the dict follow `borrow_default`.
"""
if not self.allow_short:
return False
if stock_id not in self.borrowable:
return self.borrow_default
v = self.borrowable[stock_id]
if isinstance(v, bool):
return v
try:
limit = float(v)
return amount <= limit
except Exception:
return True