129 lines
6.2 KiB
Python
129 lines
6.2 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']}")
|
||
port_after = ctx["portfolio_mv"] + add_amount
|
||
if port_after > 0 and (ctx.get("sector_mv", 0) + add_amount) / port_after > ctx["sector_max_ratio"] + 1e-9:
|
||
v.append(f"SECTOR_RATIO: 行业[{sector}]占总仓将达 "
|
||
f"{(ctx['sector_mv'] + add_amount) / port_after:.1%} > 上限 {ctx['sector_max_ratio']:.0%}")
|
||
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
|