tradingSystem/app/core/sizer.py

# -*- 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