Files
crypto_monitor/lib/trade/open_trade_gate_lib.py
dekun 7f22bffbc6 Shorten force-close window to 5m and grey-out open during blocks.
Unify Gate/OKX/Binance: disable the open button with a side note during force-close, cooloff, and daily freeze, and enforce the same gate server-side.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-06 08:38:00 +08:00

58 lines
2.0 KiB
Python

"""三所开仓门禁:账户风控 + 强制清仓窗口 + 仓位/日开仓上限."""
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,
}