126 lines
4.8 KiB
Python
126 lines
4.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
安全垫与成本账 (纯函数/纯对象, 无外部依赖, 可单测)
|
||
==================================================
|
||
职责 (POSITION_MGMT_DESIGN.md §4/§6):
|
||
1. 摊薄成本口径: avg_cost = (累计买入额 − 累计卖出额 − 做T利润) / 当前持股数。
|
||
卖得比成本高、做T赚了钱, 摊薄成本都会下降 —— 安全垫由此增厚。
|
||
2. 安全垫状态机: NONE(<0) / THIN(0~solid) / SOLID(≥solid, 解锁盈利加仓)。
|
||
3. 保垫减仓触发: 垫子峰值 ≥ peak_min 且 现值回吐 ≥ 峰值 × giveback。
|
||
4. 卖出核销次序: T0 → ADD(新→旧) → DCA → FILL → BASE (保底仓纪律)。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
|
||
class PositionCost:
|
||
"""一只股票的成本账 (摊薄口径)。所有金额单位元, 数量单位股。"""
|
||
|
||
def __init__(self):
|
||
self.qty = 0
|
||
self.cum_buy_amt = 0.0
|
||
self.cum_sell_amt = 0.0
|
||
self.t_profit = 0.0
|
||
|
||
def buy(self, qty: int, price: float):
|
||
if qty <= 0 or price <= 0:
|
||
raise ValueError(f"非法买入: qty={qty}, price={price}")
|
||
self.qty += qty
|
||
self.cum_buy_amt += qty * price
|
||
|
||
def sell(self, qty: int, price: float):
|
||
if qty <= 0 or price <= 0:
|
||
raise ValueError(f"非法卖出: qty={qty}, price={price}")
|
||
if qty > self.qty:
|
||
raise ValueError(f"卖出超持仓: sell={qty} > hold={self.qty}")
|
||
self.qty -= qty
|
||
self.cum_sell_amt += qty * price
|
||
if self.qty == 0: # 清仓即结账, 防止空仓残留成本影响下一轮
|
||
self.reset_keep_nothing()
|
||
|
||
def add_t_profit(self, amount: float):
|
||
"""做T已实现利润 (可为负 = T亏损), 直接摊入成本。"""
|
||
self.t_profit += amount
|
||
|
||
def reset_keep_nothing(self):
|
||
self.qty = 0
|
||
self.cum_buy_amt = 0.0
|
||
self.cum_sell_amt = 0.0
|
||
self.t_profit = 0.0
|
||
|
||
@property
|
||
def avg_cost(self):
|
||
"""摊薄成本。空仓返回 None; 净成本为负 (卖出已收回全部本金) 时返回 0.0 (垫子视为无限厚)。"""
|
||
if self.qty <= 0:
|
||
return None
|
||
net = self.cum_buy_amt - self.cum_sell_amt - self.t_profit
|
||
return max(net / self.qty, 0.0)
|
||
|
||
def cushion(self, price: float):
|
||
"""安全垫幅度 = 现价/摊薄成本 − 1。空仓返回 None; 成本≤0 返回大数 (视为极厚)。"""
|
||
c = self.avg_cost
|
||
if c is None:
|
||
return None
|
||
if c <= 0:
|
||
return 9.99
|
||
return price / c - 1.0
|
||
|
||
|
||
def cushion_state(cushion_pct, solid: float = 0.03) -> str:
|
||
"""垫子状态: NONE(<0) / THIN(0~solid) / SOLID(≥solid)。None 视为 NONE。"""
|
||
if cushion_pct is None or cushion_pct < 0:
|
||
return "NONE"
|
||
return "SOLID" if cushion_pct >= solid else "THIN"
|
||
|
||
|
||
def trim_trigger(cushion_peak: float, cushion_now, peak_min: float = 0.06,
|
||
giveback: float = 0.5) -> bool:
|
||
"""保垫减仓触发: 峰值曾 ≥ peak_min 且 现值回吐 ≥ 峰值 × giveback。"""
|
||
if cushion_now is None or cushion_peak is None:
|
||
return False
|
||
if cushion_peak < peak_min:
|
||
return False
|
||
return (cushion_peak - cushion_now) >= cushion_peak * giveback - 1e-12
|
||
|
||
|
||
# 卖出核销优先级 (数值小者先卖): 保底仓纪律
|
||
_SELL_ORDER = {"T0": 0, "ADD": 1, "DCA": 2, "FILL": 3, "BASE": 4, "RECON": 5}
|
||
|
||
|
||
def sell_allocation(lots: list, sell_qty: int) -> list:
|
||
"""把卖出数量分配到批次: T0 → ADD(新→旧) → DCA → FILL → BASE。
|
||
|
||
lots: [{"lot_id", "lot_type", "qty"(未核销数量), "open_date"(YYYYMMDD int 或可比较值)}, ...]
|
||
返回 [{"lot_id", "qty"}...]; 卖出量超过批次总量 → ValueError (调用方应先对账)。
|
||
"""
|
||
if sell_qty <= 0:
|
||
return []
|
||
total = sum(l["qty"] for l in lots)
|
||
if sell_qty > total:
|
||
raise ValueError(f"卖出超批次总量: sell={sell_qty} > lots={total} (先对账再动作)")
|
||
|
||
def key(l):
|
||
tier = _SELL_ORDER.get(l["lot_type"], 9)
|
||
# ADD 批内部按开仓日新→旧 (负号); 其余批按旧→新
|
||
date_key = -l["open_date"] if l["lot_type"] == "ADD" else l["open_date"]
|
||
return (tier, date_key)
|
||
|
||
out, remain = [], sell_qty
|
||
for l in sorted(lots, key=key):
|
||
if remain <= 0:
|
||
break
|
||
take = min(l["qty"], remain)
|
||
if take > 0:
|
||
out.append({"lot_id": l["lot_id"], "qty": take})
|
||
remain -= take
|
||
return out
|
||
|
||
|
||
def dca_stage(loss_pct: float, triggers=(-0.08, -0.15)) -> int:
|
||
"""补仓评估档: 返回已触及的最深档序号 (1 起), 未触及任何档返回 0。
|
||
triggers 按由浅到深排列 (如 -0.08, -0.15)。"""
|
||
stage = 0
|
||
for i, t in enumerate(sorted(triggers, reverse=True), start=1): # 浅档在前
|
||
if loss_pct <= t + 1e-12:
|
||
stage = i
|
||
return stage
|