"""三所开仓门禁:账户风控 + 强制清仓窗口 + 仓位/日开仓上限.""" from __future__ import annotations from typing import Any, Optional from lib.trade.daily_open_limit_lib import can_trade_new_open from lib.trade.force_close_lib import force_close_blocks_new_open def resolve_manual_open_gate( *, time_allows: bool, active_count: int, max_active_positions: int, opens_today: int, hard_limit: int, risk_status: Optional[dict[str, Any]], force_close_enabled: bool, force_close_bj_hour: Any, now_ms: Optional[int] = None, reset_hour: int = 8, ) -> dict[str, Any]: """汇总是否可开仓及按钮旁说明文案.""" rs = risk_status if isinstance(risk_status, dict) else {} risk_can = bool(rs.get("can_trade", True)) fc_block, fc_note = force_close_blocks_new_open( bool(force_close_enabled), force_close_bj_hour, now_ms=now_ms, ) can_trade = can_trade_new_open( time_allows=time_allows, active_count=active_count, max_active_positions=max_active_positions, opens_today=opens_today, hard_limit=hard_limit, extra_blocks=(not risk_can) or fc_block, ) note = "" if fc_block and fc_note: note = fc_note elif not risk_can: note = str(rs.get("reason") or "账户冷静期/日冻结中,暂不可开仓") elif not time_allows: note = f"未到北京时间 {int(reset_hour)}:00,暂不可开仓" elif int(active_count) >= int(max_active_positions): note = f"已达最大持仓数({int(active_count)}/{int(max_active_positions)}),暂不可开仓" elif int(hard_limit) > 0 and int(opens_today) >= int(hard_limit): note = ( f"本交易日开仓已达上限({int(opens_today)}/{int(hard_limit)})," f"次日北京时间 {int(reset_hour)}:00 后恢复" ) return { "can_trade": can_trade, "open_block_note": note if not can_trade else "", "force_close_blocks": fc_block, }