149 lines
7.8 KiB
Python
149 lines
7.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
仓位规划器 (纯函数, 无外部依赖, 可单测)
|
||
========================================
|
||
职责 (POSITION_MGMT_DESIGN.md §5):
|
||
1. 百分比制批次拆分: 单股目标仓位按 50/25/25 分批, 含一手可行性检查与自动合并
|
||
(50/25/25 → 60/40 → 100, 仍买不足一手 → 放弃并给出原因)。
|
||
2. 组合约束校验: 总仓上限 / 单股上限 / 最大持仓数 / 行业集中度 (硬拦截)。
|
||
3. 风险敞口披露: 不定量、只计算与告警。
|
||
|
||
约定: 金额单位元, 价格单位元, 数量单位股 (A股一手=100股)。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
LOT = 100 # A股一手
|
||
|
||
# 批次合并阶梯: 一手检查不过时逐级降档
|
||
MERGE_LADDER = [
|
||
("BASE/FILL/ADD", (0.5, 0.25, 0.25)),
|
||
("BASE/FILL", (0.6, 0.4)),
|
||
("BASE", (1.0,)),
|
||
]
|
||
BATCH_NAMES = ["BASE", "FILL", "ADD"]
|
||
|
||
|
||
def lot_qty(amount: float, price: float) -> int:
|
||
"""金额换算成整手股数 (向下取整到一手)。价格非法返回 0。"""
|
||
if price is None or price <= 0 or amount is None or amount <= 0:
|
||
return 0
|
||
return int(amount / price / LOT) * LOT
|
||
|
||
|
||
def split_batches(target_amount: float, price: float, splits=None, merge: bool = True) -> dict:
|
||
"""把单股目标金额拆成分批投放计划, 含一手检查与自动合并。
|
||
|
||
返回 {"ok": bool, "scheme": tuple, "batches": [{"name","amount","qty"}...], "reason": str}
|
||
- ok=False 时 batches 为空, reason 说明原因 (如"目标金额买不足一手")。
|
||
- 自动合并: 首选方案任一批次不足一手 → 逐级降档 (60/40 → 100)。
|
||
merge=False 时不合并, 直接返回失败明细 (供页面提示)。
|
||
"""
|
||
if price is None or price <= 0:
|
||
return {"ok": False, "scheme": (), "batches": [], "reason": "价格非法"}
|
||
if target_amount is None or target_amount <= 0:
|
||
return {"ok": False, "scheme": (), "batches": [], "reason": "目标金额非法"}
|
||
|
||
ladders = MERGE_LADDER if splits is None else [("CUSTOM", tuple(splits))] + (MERGE_LADDER[1:] if merge else [])
|
||
tried = []
|
||
for label, scheme in (ladders if merge else ladders[:1]):
|
||
batches = []
|
||
feasible = True
|
||
for i, ratio in enumerate(scheme):
|
||
amt = target_amount * ratio
|
||
q = lot_qty(amt, price)
|
||
if q < LOT:
|
||
feasible = False
|
||
break
|
||
name = BATCH_NAMES[i] if i < len(BATCH_NAMES) else f"B{i+1}"
|
||
batches.append({"name": name, "amount": round(amt, 2), "qty": q})
|
||
tried.append(label)
|
||
if feasible:
|
||
return {"ok": True, "scheme": scheme, "batches": batches, "reason": ""}
|
||
return {
|
||
"ok": False, "scheme": (), "batches": [],
|
||
"reason": f"目标金额 {target_amount:.0f} 元按现价 {price:.2f} 买不足一手 (已尝试: {' → '.join(tried)})",
|
||
}
|
||
|
||
|
||
def check_caps(*, ts_code: str, add_amount: float, ctx: dict) -> list:
|
||
"""组合约束硬校验。返回未通过项列表 (空列表 = 全过)。
|
||
|
||
ctx (由调用方备齐, 全部为「加仓前」快照):
|
||
scale 总规模 (元)
|
||
portfolio_cap 总仓上限 (比例)
|
||
stock_cap 单股上限 (比例)
|
||
max_names 最大持仓数
|
||
portfolio_mv 当前组合市值 (元)
|
||
names_count 当前持仓数
|
||
stock_mv 该股当前市值 (元, 无仓=0)
|
||
is_new_name 本次是否新开仓
|
||
sector 该股行业名 (None=行业数据源未配置 → 行业约束跳过)
|
||
sector_names 同行业当前持仓数
|
||
sector_mv 同行业当前市值 (元)
|
||
sector_max_names / sector_max_ratio 行业约束参数
|
||
"""
|
||
v = []
|
||
scale = float(ctx["scale"])
|
||
if scale <= 0:
|
||
return ["SCALE_INVALID: 总规模未设置"]
|
||
|
||
if (ctx["portfolio_mv"] + add_amount) / scale > ctx["portfolio_cap"] + 1e-9:
|
||
v.append(f"PORTFOLIO_CAP: 加后总仓 {(ctx['portfolio_mv'] + add_amount) / scale:.1%} "
|
||
f"> 上限 {ctx['portfolio_cap']:.0%}")
|
||
if (ctx["stock_mv"] + add_amount) / scale > ctx["stock_cap"] + 1e-9:
|
||
v.append(f"STOCK_CAP: {ts_code} 加后 {(ctx['stock_mv'] + add_amount) / scale:.1%} "
|
||
f"> 单股上限 {ctx['stock_cap']:.0%}")
|
||
if ctx.get("is_new_name") and ctx["names_count"] + 1 > ctx["max_names"]:
|
||
v.append(f"MAX_NAMES: 持仓数将达 {ctx['names_count'] + 1} > 上限 {ctx['max_names']}")
|
||
|
||
sector = ctx.get("sector")
|
||
if sector: # None/"" = 行业数据源未配置, 约束停用 (调用方负责页面提示)
|
||
if ctx.get("is_new_name") and ctx.get("sector_names", 0) + 1 > ctx["sector_max_names"]:
|
||
v.append(f"SECTOR_NAMES: 行业[{sector}]将达 {ctx['sector_names'] + 1} 只 "
|
||
f"> 上限 {ctx['sector_max_names']}")
|
||
# 行业占比: **占组合超上限 且 绝对敞口够大**, 两条同时成立才拦 (2026-07-31 定)
|
||
# ---------------------------------------------------------------------
|
||
# 只看"占组合"的话, 这道闸在建仓期是**结构性不可满足**的: 空账本买第一只票,
|
||
# 它按定义就是组合的 100%, 必然 > 任何小于 100% 的上限; 而第一只被拒之后组合市值
|
||
# 不推进, 第二只第三只面对的还是 100% —— 哪怕候选分属十个不同行业也全军覆没。
|
||
# 实测 2026-07-31: 空账本 + 10 只强传导候选 + 一条 60% 升仓命令 → 一条方案都出不来,
|
||
# 报出来却是"候选与补仓空间不足"。行业源是当天才通的 (此前 sector 恒为 None、整段
|
||
# 跳过), 所以这个洞一直藏着。数学上上限 40% 至少要 3 只不同行业的票同时在组合里才
|
||
# 可能满足, 调松阈值解决不了。
|
||
#
|
||
# 加"绝对敞口"这条判据的道理: 集中度是风险的**放大器**, 不是风险本身 —— 敞口只有
|
||
# 规模 6% 的时候, 它 100% 集中在一个行业也谈不上风险。绝对线取
|
||
# `sector_max_ratio × portfolio_cap`, 于是组合建满到总仓上限时两条判据自然趋同
|
||
# (40% × 60% = 24% of scale ⟺ 40% of portfolio), 中间是连续过渡, 没有"第几只突然
|
||
# 开始生效"的台阶。
|
||
port_after = ctx["portfolio_mv"] + add_amount
|
||
sector_after = ctx.get("sector_mv", 0) + add_amount
|
||
ratio_max = ctx["sector_max_ratio"]
|
||
of_port = (sector_after / port_after) if port_after > 0 else 0.0
|
||
of_scale = sector_after / scale
|
||
floor = ratio_max * float(ctx.get("portfolio_cap") or 1.0)
|
||
if of_port > ratio_max + 1e-9 and of_scale > floor + 1e-9:
|
||
v.append(f"SECTOR_RATIO: 行业[{sector}]占组合将达 {of_port:.1%} > 上限 "
|
||
f"{ratio_max:.0%}, 且占规模 {of_scale:.1%} > 绝对线 {floor:.1%}")
|
||
return v
|
||
|
||
|
||
def risk_exposure(qty: int, price: float, stop_ref: float) -> float:
|
||
"""单笔风险敞口 (元) = 数量 × max(0, 买价 − 止损参考)。stop_ref 缺失返回 -1 表示无法计算。"""
|
||
if stop_ref is None or stop_ref <= 0:
|
||
return -1.0
|
||
return qty * max(0.0, price - stop_ref)
|
||
|
||
|
||
def risk_warnings(*, entry_exposure: float, portfolio_exposure: float, scale: float,
|
||
warn_entry: float = 0.01, warn_portfolio: float = 0.06) -> list:
|
||
"""风险披露告警 (不拦截, 只提示)。entry_exposure=-1 时提示无法计算。"""
|
||
w = []
|
||
if entry_exposure < 0:
|
||
w.append("RISK_UNKNOWN: 无止损参考位, 敞口无法计算")
|
||
elif scale > 0 and entry_exposure / scale > warn_entry:
|
||
w.append(f"RISK_ENTRY: 单笔敞口 {entry_exposure / scale:.2%} > 披露线 {warn_entry:.0%}")
|
||
if scale > 0 and portfolio_exposure > 0 and portfolio_exposure / scale > warn_portfolio:
|
||
w.append(f"RISK_PORTFOLIO: 组合敞口 {portfolio_exposure / scale:.2%} > 披露线 {warn_portfolio:.0%}")
|
||
return w
|