计仓与保本说明
diff --git a/lib/trade/force_close_lib.py b/lib/trade/force_close_lib.py
index 46165e0..ef7c1db 100644
--- a/lib/trade/force_close_lib.py
+++ b/lib/trade/force_close_lib.py
@@ -8,7 +8,22 @@ from typing import Any, Optional
from zoneinfo import ZoneInfo
FORCE_CLOSE_RESULT = "强制清仓"
-FORCE_CLOSE_GRACE_MINUTES = 15
+# 默认宽限(分钟);运行时优先读 FORCE_CLOSE_GRACE_MINUTES
+FORCE_CLOSE_GRACE_MINUTES = 5
+
+
+def force_close_grace_minutes(override: Any = None) -> int:
+ """整点强制清仓执行窗口长度(分钟)."""
+ if override is not None and str(override).strip() != "":
+ raw = override
+ else:
+ raw = os.getenv("FORCE_CLOSE_GRACE_MINUTES")
+ if raw is None or str(raw).strip() == "":
+ raw = FORCE_CLOSE_GRACE_MINUTES
+ try:
+ return max(1, int(raw))
+ except (TypeError, ValueError):
+ return max(1, int(FORCE_CLOSE_GRACE_MINUTES))
def app_timezone_name() -> str:
@@ -43,7 +58,7 @@ def is_force_close_active_hour(
*,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
- grace_minutes: int = FORCE_CLOSE_GRACE_MINUTES,
+ grace_minutes: Optional[int] = None,
) -> bool:
"""当前是否处于整点强制清仓执行窗口(整点起 grace 分钟内)."""
return is_force_close_executing(
@@ -59,17 +74,41 @@ def is_force_close_executing(
*,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
- grace_minutes: int = FORCE_CLOSE_GRACE_MINUTES,
+ grace_minutes: Optional[int] = None,
) -> bool:
hour = normalize_force_close_bj_hour(bj_hour)
now = _now_dt(now_ms=now_ms, tz_name=tz_name)
target = now.replace(hour=hour, minute=0, second=0, microsecond=0)
if now < target:
return False
- end = target + timedelta(minutes=max(1, int(grace_minutes)))
+ grace = force_close_grace_minutes(grace_minutes)
+ end = target + timedelta(minutes=grace)
return now < end
+def force_close_blocks_new_open(
+ enabled: bool,
+ bj_hour: Any,
+ *,
+ now_ms: Optional[int] = None,
+ tz_name: Optional[str] = None,
+ grace_minutes: Optional[int] = None,
+) -> tuple[bool, str]:
+ """强制清仓执行窗口内禁止新开仓.返回 (是否拦截, 说明文案)."""
+ if not enabled:
+ return False, ""
+ grace = force_close_grace_minutes(grace_minutes)
+ if not is_force_close_executing(
+ bj_hour, now_ms=now_ms, tz_name=tz_name, grace_minutes=grace
+ ):
+ return False, ""
+ label = force_close_hour_label(bj_hour)
+ return (
+ True,
+ f"强制清仓窗口内(北京时间 {label} 起 {grace} 分钟),暂不可开仓",
+ )
+
+
def parse_closed_at_dt(
closed_at: Any,
*,
@@ -93,7 +132,7 @@ def is_close_at_force_close_window(
closed_at: Any,
bj_hour: Any,
*,
- grace_minutes: int = FORCE_CLOSE_GRACE_MINUTES,
+ grace_minutes: Optional[int] = None,
tz_name: Optional[str] = None,
) -> bool:
"""平仓时刻是否落在北京时间整点强制清仓窗口内."""
@@ -103,7 +142,7 @@ def is_close_at_force_close_window(
hour = normalize_force_close_bj_hour(bj_hour)
if dt.hour != hour:
return False
- return dt.minute < max(1, int(grace_minutes))
+ return dt.minute < force_close_grace_minutes(grace_minutes)
def infer_force_close_result(
@@ -111,7 +150,7 @@ def infer_force_close_result(
*,
enabled: bool,
bj_hour: Any,
- grace_minutes: int = FORCE_CLOSE_GRACE_MINUTES,
+ grace_minutes: Optional[int] = None,
tz_name: Optional[str] = None,
) -> Optional[str]:
if not enabled:
@@ -130,7 +169,7 @@ def coerce_force_close_result(
enabled: bool,
bj_hour: Any,
miss_reason: Optional[str] = None,
- grace_minutes: int = FORCE_CLOSE_GRACE_MINUTES,
+ grace_minutes: Optional[int] = None,
tz_name: Optional[str] = None,
) -> tuple[str, str]:
"""同步平仓归类:整点窗口内优先记为强制清仓."""
@@ -158,7 +197,7 @@ def apply_force_close_display_result(
*,
enabled: bool,
bj_hour: Any,
- grace_minutes: int = FORCE_CLOSE_GRACE_MINUTES,
+ grace_minutes: Optional[int] = None,
tz_name: Optional[str] = None,
) -> str:
"""展示层:外部平仓/手动平仓若落在整点窗口,显示为强制清仓."""
@@ -227,19 +266,24 @@ def build_force_close_state(
has_active_positions: Optional[bool] = None,
) -> dict[str, Any]:
"""实例级强制清仓状态(模板 / API 共用)."""
+ grace = force_close_grace_minutes()
if not enabled:
return {
"enabled": False,
"bj_hour": normalize_force_close_bj_hour(bj_hour),
"hour_label": force_close_hour_label(bj_hour),
"label": force_close_label(bj_hour),
+ "grace_minutes": grace,
"next_at_ms": None,
"remaining_sec": None,
"countdown": "",
"active": False,
+ "executing": False,
}
hour = normalize_force_close_bj_hour(bj_hour)
- executing = is_force_close_executing(hour, now_ms=now_ms, tz_name=tz_name)
+ executing = is_force_close_executing(
+ hour, now_ms=now_ms, tz_name=tz_name, grace_minutes=grace
+ )
active = executing and (has_active_positions is not False)
next_at_ms = compute_next_force_close_at_ms(bj_hour=hour, now_ms=now_ms, tz_name=tz_name)
rem = force_close_remaining_seconds(next_at_ms, now_ms=now_ms) if next_at_ms else None
@@ -248,10 +292,12 @@ def build_force_close_state(
"bj_hour": hour,
"hour_label": force_close_hour_label(hour),
"label": force_close_label(hour),
+ "grace_minutes": grace,
"next_at_ms": next_at_ms,
"remaining_sec": rem,
"countdown": format_force_close_countdown(rem, active=active),
"active": active,
+ "executing": executing,
}
diff --git a/lib/trade/open_trade_gate_lib.py b/lib/trade/open_trade_gate_lib.py
new file mode 100644
index 0000000..be298d5
--- /dev/null
+++ b/lib/trade/open_trade_gate_lib.py
@@ -0,0 +1,57 @@
+"""三所开仓门禁:账户风控 + 强制清仓窗口 + 仓位/日开仓上限."""
+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,
+ }
diff --git a/tests/test_force_close_lib.py b/tests/test_force_close_lib.py
index 54d3397..6555a33 100644
--- a/tests/test_force_close_lib.py
+++ b/tests/test_force_close_lib.py
@@ -49,7 +49,8 @@ def test_next_force_close_after_trigger_same_day():
def test_executing_window_and_countdown():
- now = _ms(2026, 7, 6, 0, 14)
+ # 默认宽限 5 分钟:00:04 仍在执行窗
+ now = _ms(2026, 7, 6, 0, 4)
assert is_force_close_executing(0, now_ms=now, tz_name="Asia/Shanghai")
assert is_force_close_active_hour(0, now_ms=now, tz_name="Asia/Shanghai")
state = build_force_close_state(
@@ -57,17 +58,21 @@ def test_executing_window_and_countdown():
)
assert state["enabled"] is True
assert state["active"] is True
+ assert state["executing"] is True
+ assert state["grace_minutes"] == 5
assert state["countdown"] == "执行中"
assert state["next_at_ms"] == _ms(2026, 7, 7, 0, 0)
def test_not_executing_after_grace_without_positions():
- now = _ms(2026, 7, 7, 0, 18)
+ # 默认宽限 5 分钟:00:05 起已离开执行窗
+ now = _ms(2026, 7, 7, 0, 5)
assert not is_force_close_executing(0, now_ms=now, tz_name="Asia/Shanghai")
state = build_force_close_state(
True, 0, now_ms=now, tz_name="Asia/Shanghai", has_active_positions=False
)
assert state["active"] is False
+ assert state["executing"] is False
assert state["countdown"] != "执行中"
assert state["next_at_ms"] == _ms(2026, 7, 8, 0, 0)
@@ -85,6 +90,8 @@ def test_format_countdown():
def test_infer_force_close_from_closed_at():
assert is_close_at_force_close_window("2026-07-07 00:00", 0)
+ assert is_close_at_force_close_window("2026-07-07 00:04", 0)
+ assert not is_close_at_force_close_window("2026-07-07 00:05", 0)
assert infer_force_close_result("2026-07-07 00:00", enabled=True, bj_hour=0) == "强制清仓"
assert infer_force_close_result("2026-07-07 00:20", enabled=True, bj_hour=0) is None
diff --git a/tests/test_open_trade_gate_lib.py b/tests/test_open_trade_gate_lib.py
new file mode 100644
index 0000000..ca137c2
--- /dev/null
+++ b/tests/test_open_trade_gate_lib.py
@@ -0,0 +1,43 @@
+from lib.trade.open_trade_gate_lib import resolve_manual_open_gate
+
+
+def _base(**kwargs):
+ cfg = dict(
+ time_allows=True,
+ active_count=0,
+ max_active_positions=1,
+ opens_today=0,
+ hard_limit=10,
+ risk_status={"can_trade": True, "reason": ""},
+ force_close_enabled=False,
+ force_close_bj_hour=0,
+ now_ms=None,
+ reset_hour=8,
+ )
+ cfg.update(kwargs)
+ return resolve_manual_open_gate(**cfg)
+
+
+def test_open_when_clear():
+ g = _base()
+ assert g["can_trade"] is True
+ assert g["open_block_note"] == ""
+
+
+def test_block_cooloff_note():
+ g = _base(risk_status={"can_trade": False, "reason": "账户冷静期中"})
+ assert g["can_trade"] is False
+ assert "冷静期" in g["open_block_note"]
+
+
+def test_block_force_close_window():
+ # 2026-07-07 00:02 Asia/Shanghai
+ from datetime import datetime
+ from zoneinfo import ZoneInfo
+
+ now_ms = int(datetime(2026, 7, 7, 0, 2, tzinfo=ZoneInfo("Asia/Shanghai")).timestamp() * 1000)
+ g = _base(force_close_enabled=True, force_close_bj_hour=0, now_ms=now_ms)
+ assert g["can_trade"] is False
+ assert g["force_close_blocks"] is True
+ assert "强制清仓" in g["open_block_note"]
+ assert "5 分钟" in g["open_block_note"]