"""交易上下文:设置读取、资金、模式。""" from __future__ import annotations from typing import Callable, Optional TRADING_MODE_SIM = "simulation" # SimNow CTP TRADING_MODE_LIVE = "live" # 期货公司 CTP def get_trading_mode(get_setting: Callable[[str, str], str]) -> str: m = (get_setting("trading_mode", TRADING_MODE_SIM) or TRADING_MODE_SIM).strip().lower() return m if m in (TRADING_MODE_SIM, TRADING_MODE_LIVE) else TRADING_MODE_SIM def get_sizing_mode(get_setting: Callable[[str, str], str]) -> str: from position_sizing import normalize_sizing_mode return normalize_sizing_mode(get_setting("position_sizing_mode", "fixed")) def get_fixed_lots(get_setting: Callable[[str, str], str]) -> int: try: return max(1, int(float(get_setting("fixed_lots", "1") or 1))) except (TypeError, ValueError): return 1 def get_fixed_amount(get_setting: Callable[[str, str], str]) -> float: try: return max(1.0, float(get_setting("fixed_amount", "5000") or 5000)) except (TypeError, ValueError): return 5000.0 def get_risk_percent(get_setting: Callable[[str, str], str]) -> float: try: return max(0.1, float(get_setting("risk_percent", "1") or 1)) except (TypeError, ValueError): return 1.0 def get_max_margin_pct(get_setting: Callable[[str, str], str]) -> float: """单笔/总仓位保证金占权益上限(%),默认 30。""" try: return max(1.0, min(100.0, float(get_setting("max_margin_pct", "30") or 30))) except (TypeError, ValueError): return 30.0 def get_trailing_be_tick_buffer(get_setting: Callable[[str, str], str]) -> int: """移动保本:止损移至开仓价 ± N 个最小变动价位(默认 2)。""" try: return max(1, min(20, int(float(get_setting("trailing_be_tick_buffer", "2") or 2)))) except (TypeError, ValueError): return 2 def get_pending_order_timeout_min(get_setting: Callable[[str, str], str]) -> int: """开仓限价委托未成交自动撤单时间(分钟),默认 5。""" try: return max(1, min(60, int(float(get_setting("pending_order_timeout_min", "5") or 5)))) except (TypeError, ValueError): return 5 def get_pending_order_timeout_sec(get_setting: Callable[[str, str], str]) -> int: return get_pending_order_timeout_min(get_setting) * 60 def get_account_capital(conn, get_setting: Callable[[str, str], str]) -> float: """优先 SimNow/期货公司 CTP 权益;未连接时用设置中的参考资金。""" del conn mode = get_trading_mode(get_setting) try: from vnpy_bridge import ctp_status, get_ctp_balance st = ctp_status(mode) if st.get("connected"): bal = get_ctp_balance(mode) if bal and bal > 0: return float(bal) except Exception: pass try: return float(get_setting("live_capital", "0") or 0) except (TypeError, ValueError): return 0.0 def trading_mode_label(get_setting: Callable[[str, str], str]) -> str: return "SimNow" if get_trading_mode(get_setting) == TRADING_MODE_SIM else "期货公司实盘"