91 lines
3.8 KiB
Python
91 lines
3.8 KiB
Python
"""ZT 正T/反T 的触发与数量规则：纯函数，不碰网络、存储和线程。
|
||
|
||
设计要点：
|
||
|
||
* 以建仓价 ``base_cost`` 为中枢，中性带 ``zt_t_band_pct`` 内不做任何动作。
|
||
现价低于下沿只考虑正T，高于上沿只考虑反T —— 同一时刻只可能命中一种方向，
|
||
天然满足"一只股票每天只做一轮"，不需要额外的冲突仲裁。
|
||
* 数量一律再受券商现实约束封顶：卖出封顶 ``can_use_volume``（T+1 只在这里
|
||
体现），买入封顶可用资金。正T 当天买入的份额当天不可卖，因此它的平仓腿
|
||
能卖多少完全由 ``can_use_volume`` 决定，卖不动就自然留成隔夜持仓。
|
||
"""
|
||
|
||
from .rounds import KIND_LONG_T, KIND_SHORT_T
|
||
|
||
LOT = 100
|
||
|
||
|
||
def choose_kind(price: float, base_cost: float, band_pct: float) -> str | None:
|
||
"""按现价相对建仓价的位置决定本轮方向；中性带内返回 None。"""
|
||
if price <= 0 or base_cost <= 0 or band_pct < 0:
|
||
return None
|
||
if price <= base_cost * (1 - band_pct / 100):
|
||
return KIND_LONG_T
|
||
if price >= base_cost * (1 + band_pct / 100):
|
||
return KIND_SHORT_T
|
||
return None
|
||
|
||
|
||
def price_allowed(price: float, max_price: float) -> bool:
|
||
"""高价股不参与做 T。"""
|
||
return 0 < price <= max_price
|
||
|
||
|
||
def entry_volume(kind: str, *, price: float, open_hands: int, sell_ratio: float,
|
||
base_qty: int, can_use_volume: int, available: float) -> int:
|
||
"""开仓腿计划数量；0 表示不提交。"""
|
||
if price <= 0:
|
||
return 0
|
||
if kind == KIND_LONG_T:
|
||
# 正T 买入：按手数取量，再受可用资金封顶。
|
||
affordable = int(max(0.0, available) // (price * LOT)) * LOT
|
||
return max(0, min(open_hands * LOT, affordable))
|
||
if kind == KIND_SHORT_T:
|
||
# 反T 卖出：按建仓数量比例取整手，再受可卖库存封顶。
|
||
planned = int(base_qty * sell_ratio) // LOT * LOT
|
||
return max(0, min(planned, _whole_lots(can_use_volume)))
|
||
return 0
|
||
|
||
|
||
def exit_volume(kind: str, *, residual_qty: int, price: float,
|
||
can_use_volume: int, available: float) -> int:
|
||
"""平仓腿可提交数量；0 表示当前无法平仓（T+1 冻结或资金不足）。"""
|
||
if residual_qty <= 0 or price <= 0:
|
||
return 0
|
||
if kind == KIND_LONG_T:
|
||
return max(0, min(residual_qty, _whole_lots(can_use_volume)))
|
||
if kind == KIND_SHORT_T:
|
||
affordable = int(max(0.0, available) // (price * LOT)) * LOT
|
||
return max(0, min(residual_qty, affordable))
|
||
return 0
|
||
|
||
|
||
def entry_triggered(kind: str, price: float, base_cost: float, *, band_pct: float,
|
||
rebound_confirmed: bool, retrace_confirmed: bool) -> bool:
|
||
"""开仓腿是否满足触发条件。"""
|
||
if choose_kind(price, base_cost, band_pct) != kind:
|
||
return False
|
||
if kind == KIND_LONG_T:
|
||
return rebound_confirmed # 低吸要等反弹确认，不接下跌中的飞刀
|
||
return retrace_confirmed # 高抛要等盈利网格回撤，不追最高点
|
||
|
||
|
||
def exit_triggered(kind: str, price: float, entry_avg_price: float, *,
|
||
buy_fall_pct: float, profit_step_pct: float,
|
||
rebound_confirmed: bool) -> bool:
|
||
"""平仓腿是否满足触发条件。"""
|
||
if entry_avg_price <= 0:
|
||
return False
|
||
if kind == KIND_SHORT_T:
|
||
# 反T 买回：较卖出均价回落 buy_fall_pct 且已见反弹。
|
||
target = entry_avg_price * (1 - buy_fall_pct / 100)
|
||
return price <= target and rebound_confirmed
|
||
if kind == KIND_LONG_T:
|
||
# 正T 卖出：较买入均价上涨一个网格步长。
|
||
return price >= entry_avg_price * (1 + profit_step_pct / 100)
|
||
return False
|
||
|
||
|
||
def _whole_lots(volume: int) -> int:
|
||
return max(0, int(volume)) // LOT * LOT
|