"""Portfolio Engine(v2 §16)—— 组合构建模块(M8.2)。 MVP:等权资金拆分(与既有 TopK 回测等权语义一致,行为收敛到本模块); 单股/行业上限等约束为预留字段,未建模时由回测器写入 unimplemented (禁止假装支持,AGENT.md §24)。 """ from __future__ import annotations from app.domain.entities.research import PortfolioSpec def equal_weight_budget(cash: float, target_count: int) -> float: """等权单标的预算:现金均分(target_count>0)。""" if target_count <= 0: return 0.0 return cash / target_count def unimplemented_notes(portfolio: PortfolioSpec) -> list[str]: """组合层未建模项说明(默认空;设置约束即显式标注)。 max_position_pct 已建模(Portfolio v1.1 单股上限分配);行业上限依赖行业元数据 注入(v1.1 起仍标注未建模,禁止假装支持)。 """ notes: list[str] = [] if portfolio.max_industry_weight_pct is not None: notes.append( f"最大行业权重 {portfolio.max_industry_weight_pct:.0%} 约束未建模(需行业元数据注入)" ) return notes def allocate_with_max_position( cash: float, targets: list[str], equity: float, max_position_pct: float | None, ) -> dict[str, float]: """按单股上限的等权资金分配(M9/C2,Portfolio v1.1)。 - 无上限 → 现金均分(与原等权语义一致) - 有上限 cap:迭代把「均分份额超过 cap×equity」的标的封顶,剩余现金在其余标的中 继续均分,直至收敛;未分配现金留在组合(现金管理)。 """ n = len(targets) if n == 0: return {} if max_position_pct is None: return {t: cash / n for t in targets} cap_value = max_position_pct * equity if cap_value <= 0: return {t: 0.0 for t in targets} spend: dict[str, float] = {} left = cash pool = list(targets) while pool and left > 1e-6: share = left / len(pool) capped = [t for t in pool if share > cap_value + 1e-9] if not capped: for t in pool: spend[t] = share break for t in capped: spend[t] = cap_value left -= cap_value pool = [x for x in pool if x != t] for t in targets: spend.setdefault(t, 0.0) return spend