diff --git a/crypto_monitor_binance/.env.example b/crypto_monitor_binance/.env.example index ae546a6..d6294e7 100644 --- a/crypto_monitor_binance/.env.example +++ b/crypto_monitor_binance/.env.example @@ -52,6 +52,13 @@ UPLOAD_DIR=static/images # BINANCE_FUNDING_INCLUDE_SPOT=false # 计仓：risk=以损定仓（默认）；full_margin=合约可用×FULL_MARGIN_BUFFER_RATIO 全仓杠杆（须无仓后重启） POSITION_SIZING_MODE=risk +# 方向限制（默认 false=双向均可；true 时按 TRADE_DIRECTION 限制，修改后须重启） +# TRADE_DIRECTION=long_only | short_only | both（或 多/空/双向） +TRADE_DIRECTION_RESTRICT_ENABLED=false +TRADE_DIRECTION=both +# 币种白名单（默认 false=全币种可手输；true 时关键位/下单/策略仅下拉选择） +TRADE_SYMBOL_RESTRICT_ENABLED=false +TRADE_SYMBOL_WHITELIST=BTC,ETH # 每天起始基数（U） DAILY_START_CAPITAL=30 # 日内回撤后基数（U） diff --git a/crypto_monitor_binance/app.py b/crypto_monitor_binance/app.py index 31b40ca..e12caa2 100644 --- a/crypto_monitor_binance/app.py +++ b/crypto_monitor_binance/app.py @@ -159,6 +159,14 @@ from lib.trade.position_sizing_lib import ( mode_label_zh, risk_percent_for_storage, ) +from lib.trade.trade_policy_lib import load_trade_policy +from lib.trade.trade_policy_app_lib import ( + check_direction_policy, + check_open_policy, + check_symbol_policy, + default_symbol_for_policy, + trade_policy_template_context, +) from lib.key_monitor.key_monitor_full_margin_lib import ( monitor_type_disallowed_in_full_margin, purge_disallowed_key_monitors, @@ -342,6 +350,7 @@ FORCE_CLOSE_BJ_HOUR = int(os.getenv("FORCE_CLOSE_BJ_HOUR", "0")) AUTO_TRANSFER_BJ_HOUR = int(os.getenv("AUTO_TRANSFER_BJ_HOUR", "8")) # 计仓模式：risk=以损定仓（默认）；full_margin=合约可用保证金×比例全仓杠杆（仅 env 切换，须无仓） POSITION_SIZING_MODE = load_position_sizing_mode() +TRADE_POLICY = load_trade_policy() WECHAT_TIMEOUT_SECONDS = int(os.getenv("WECHAT_TIMEOUT_SECONDS", "10")) AI_TIMEOUT_SECONDS = int(os.getenv("AI_TIMEOUT_SECONDS", "120")) MONITOR_POLL_SECONDS = int(os.getenv("MONITOR_POLL_SECONDS", "3")) @@ -2035,6 +2044,12 @@ def normalize_symbol_input(symbol): return f"{sym}/USDT" +def validate_trade_policy_open(symbol, direction): + return check_open_policy( + TRADE_POLICY, symbol, direction, normalize_symbol_input + ) + + def normalize_kline_limit(limit_raw, default=200): try: n = int(limit_raw) @@ -7277,6 +7292,7 @@ def render_main_page(page="trade", embed_mode=None): risk_percent=RISK_PERCENT, position_sizing_mode=POSITION_SIZING_MODE, position_sizing_mode_label=mode_label_zh(POSITION_SIZING_MODE), + trade_policy=trade_policy_template_context(TRADE_POLICY), open_position_button_label=( "开仓（全仓杠杆）" if is_full_margin_mode(POSITION_SIZING_MODE) else "开仓（以损定仓）" ), @@ -7996,7 +8012,10 @@ def key_focus(): selected_key = next((k for k in key_list if (k.get("symbol") or "").upper() == symbol_query), None) if selected_key is None and key_list: selected_key = key_list[0] - default_symbol = symbol_query or ((selected_key or {}).get("symbol")) or "BTC/USDT" + default_symbol = default_symbol_for_policy( + TRADE_POLICY, + symbol_query or ((selected_key or {}).get("symbol")) or "BTC/USDT", + ) return render_template( "key_focus_v2.html", key_list=key_list, @@ -8006,6 +8025,7 @@ def key_focus(): default_kline_limit=200, price_refresh_seconds=PRICE_REFRESH_SECONDS, exchange_display=EXCHANGE_DISPLAY_NAME, + trade_policy=trade_policy_template_context(TRADE_POLICY), ) @@ -8116,6 +8136,12 @@ def add_key(): if not symbol: flash("symbol 不能为空") return redirect("/key_monitor") + ok_sym, sym_msg = check_symbol_policy( + TRADE_POLICY, symbol, normalize_symbol_input + ) + if not ok_sym: + flash(sym_msg) + return redirect("/key_monitor") mt = (d.get("type") or "").strip() direction_sel = (d.get("direction") or "").strip().lower() dup_msg = check_duplicate_submit( @@ -8130,6 +8156,10 @@ def add_key(): elif direction_sel not in ("long", "short"): flash("箱体/收敛突破请选择做多或做空") return redirect("/key_monitor") + ok_dir, dir_msg = check_direction_policy(TRADE_POLICY, direction_sel) + if not ok_dir: + flash(dir_msg) + return redirect("/key_monitor") allowed_types = ( tuple(KEY_MONITOR_AUTO_TYPES) + tuple(KEY_MONITOR_ALERT_ONLY_TYPES) @@ -8375,6 +8405,11 @@ def add_order(): conn.close() flash("symbol 不能为空") return redirect("/") + ok_pol, pol_msg = validate_trade_policy_open(symbol, direction) + if not ok_pol: + conn.close() + flash(f"账户限制：{pol_msg}") + return redirect("/trade") dup_msg = check_duplicate_submit(session, submit_scope_add_order(symbol, direction)) if dup_msg: conn.close() @@ -9703,6 +9738,7 @@ def _hub_meta_bundle(): "max_active_positions": MAX_ACTIVE_POSITIONS, "btc_leverage": BTC_LEVERAGE, "alt_leverage": ALT_LEVERAGE, + "trade_policy": trade_policy_template_context(TRADE_POLICY), } diff --git a/crypto_monitor_binance/templates/index.html b/crypto_monitor_binance/templates/index.html index b27d408..e6c487a 100644 --- a/crypto_monitor_binance/templates/index.html +++ b/crypto_monitor_binance/templates/index.html @@ -243,7 +243,7 @@ .stats-period-block h3{font-size:1rem;color:#dbe4ff;margin-bottom:4px} .stats-period-block .sub{font-size:.78rem;color:#8892b0;margin-bottom:10px;line-height:1.4} - + 加密货币｜交易监控 + AI复盘一体化
{{ exchange_display }}
+ {% if trade_policy.badge_text %} + {{ trade_policy.badge_text }} + {% endif %} {{ risk_status.status_label|default('正常') }}
- - + {% from 'trade_policy_fields.html' import trade_policy_symbol, trade_policy_direction %} + {{ trade_policy_symbol('symbol', 'order-symbol') }} + {{ trade_policy_direction('direction', 'order-direction') }} - + {% from 'trade_policy_fields.html' import trade_policy_symbol, trade_policy_direction %} + {{ trade_policy_symbol('symbol', 'order-symbol') }} + {{ trade_policy_direction('direction', 'order-direction') }} - + {% from 'trade_policy_fields.html' import trade_policy_symbol, trade_policy_direction %} + {{ trade_policy_symbol('symbol', 'order-symbol') }} + {{ trade_policy_direction('direction', 'order-direction') }} - + {% from 'trade_policy_fields.html' import trade_policy_symbol, trade_policy_direction %} + {{ trade_policy_symbol('symbol', 'order-symbol') }} + {{ trade_policy_direction('direction', 'order-direction') }} + {% from 'trade_policy_fields.html' import trade_policy_symbol %} + {{ trade_policy_symbol('symbol', 'symbol-input', default_symbol, placeholder='BTC/USDT') }} + {% from 'trade_policy_fields.html' import trade_policy_symbol, trade_policy_direction %} + {{ trade_policy_symbol('symbol', 'key-symbol') }} - + {{ trade_policy_direction('direction', 'key-direction') }} diff --git a/lib/strategy/templates/strategy_trend_panel.html b/lib/strategy/templates/strategy_trend_panel.html index 64e8347..7e5271a 100644 --- a/lib/strategy/templates/strategy_trend_panel.html +++ b/lib/strategy/templates/strategy_trend_panel.html @@ -24,12 +24,9 @@ {% endfor %} {% endif %} - - + {% from 'trade_policy_fields.html' import trade_policy_symbol, trade_policy_direction %} + {{ trade_policy_symbol('symbol', 'trend-symbol', placeholder='BTC 或 ETH/USDT') }} + {{ trade_policy_direction('direction', 'trend-direction') }} diff --git a/lib/strategy/templates/trade_policy_fields.html b/lib/strategy/templates/trade_policy_fields.html new file mode 100644 index 0000000..fe63bd3 --- /dev/null +++ b/lib/strategy/templates/trade_policy_fields.html @@ -0,0 +1,29 @@ +{# 方向 / 币种：env 账户级限制（三所共用宏） #} +{% macro trade_policy_symbol(name, id, value='', required=true, placeholder='BTC 或 BTC/USDT') -%} +{% if trade_policy.symbol_restrict_enabled and trade_policy.symbol_whitelist %} + +{% else %} + +{% endif %} +{%- endmacro %} + +{% macro trade_policy_direction(name, id, required=true, include_empty=true) -%} +{% if trade_policy.direction_restrict_enabled and trade_policy.direction_mode == 'long_only' %} +做多 + +{% elif trade_policy.direction_restrict_enabled and trade_policy.direction_mode == 'short_only' %} +做空 + +{% else %} + +{% endif %} +{%- endmacro %} diff --git a/lib/trade/trade_policy_app_lib.py b/lib/trade/trade_policy_app_lib.py new file mode 100644 index 0000000..7b9f4af --- /dev/null +++ b/lib/trade/trade_policy_app_lib.py @@ -0,0 +1,52 @@ +"""Flask 实例接入 trade policy（三所 app.py 共用）。""" +from __future__ import annotations + +from typing import Callable, Tuple + +from lib.trade.trade_policy_lib import ( + TradePolicy, + assert_direction_allowed, + assert_symbol_allowed, + assert_trade_policy_open, + trade_policy_to_dict, +) + + +def trade_policy_template_context(policy: TradePolicy) -> dict: + return trade_policy_to_dict(policy) + + +def default_symbol_for_policy(policy: TradePolicy, raw_default: str) -> str: + d = (raw_default or "BTC/USDT").strip() or "BTC/USDT" + if policy.symbol_restrict_enabled and policy.symbol_whitelist: + from lib.trade.trade_policy_lib import symbol_base_coin + + base = symbol_base_coin(d) + if base not in policy.symbol_whitelist: + return f"{policy.symbol_whitelist[0]}/USDT" + return d + + +def check_symbol_policy( + policy: TradePolicy, + symbol: str, + normalize_symbol_fn: Callable[[str], str], +) -> Tuple[bool, str]: + return assert_symbol_allowed( + policy, symbol, normalize_symbol_fn=normalize_symbol_fn + ) + + +def check_direction_policy(policy: TradePolicy, direction: str) -> Tuple[bool, str]: + return assert_direction_allowed(policy, direction) + + +def check_open_policy( + policy: TradePolicy, + symbol: str, + direction: str, + normalize_symbol_fn: Callable[[str], str], +) -> Tuple[bool, str]: + return assert_trade_policy_open( + policy, symbol, direction, normalize_symbol_fn=normalize_symbol_fn + ) diff --git a/lib/trade/trade_policy_lib.py b/lib/trade/trade_policy_lib.py new file mode 100644 index 0000000..60b25c0 --- /dev/null +++ b/lib/trade/trade_policy_lib.py @@ -0,0 +1,205 @@ +""" +三所共用：账户级方向 / 币种白名单（.env 开关，默认关闭=不限制）。 +""" +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Callable, FrozenSet, Optional, Sequence, Tuple + +DIR_BOTH = "both" +DIR_LONG_ONLY = "long_only" +DIR_SHORT_ONLY = "short_only" +VALID_DIRECTION_MODES = frozenset({DIR_BOTH, DIR_LONG_ONLY, DIR_SHORT_ONLY}) + +_DIR_ALIASES = { + "both": DIR_BOTH, + "双向": DIR_BOTH, + "long": DIR_LONG_ONLY, + "long_only": DIR_LONG_ONLY, + "多": DIR_LONG_ONLY, + "仅多": DIR_LONG_ONLY, + "做多": DIR_LONG_ONLY, + "short": DIR_SHORT_ONLY, + "short_only": DIR_SHORT_ONLY, + "空": DIR_SHORT_ONLY, + "仅空": DIR_SHORT_ONLY, + "做空": DIR_SHORT_ONLY, +} + + +def _env_bool(raw: Optional[str], default: bool = False) -> bool: + if raw is None: + return default + return (raw or "").strip().lower() in ("1", "true", "yes", "on") + + +def normalize_direction_mode(raw: Optional[str]) -> str: + v = (raw or DIR_BOTH).strip().lower() + return _DIR_ALIASES.get(v, v if v in VALID_DIRECTION_MODES else DIR_BOTH) + + +def symbol_base_coin(symbol: str) -> str: + """BTC/USDT:USDT、BTC/USDT、BTC、btc -> BTC""" + s = (symbol or "").strip().upper() + if not s: + return "" + if ":" in s: + s = s.split(":", 1)[0] + if "/" in s: + return s.split("/", 1)[0].strip() + if s.endswith("USDT") and len(s) > 4: + return s[:-4] + return s + + +def parse_symbol_whitelist(raw: Optional[str]) -> Tuple[str, ...]: + if not raw or not str(raw).strip(): + return () + parts = [] + for piece in str(raw).replace(";", ",").split(","): + base = symbol_base_coin(piece.strip()) + if base and base not in parts: + parts.append(base) + return tuple(parts) + + +@dataclass(frozen=True) +class TradePolicy: + direction_restrict_enabled: bool + direction_mode: str + symbol_restrict_enabled: bool + symbol_whitelist: Tuple[str, ...] + + @property + def allows_long(self) -> bool: + if not self.direction_restrict_enabled: + return True + return self.direction_mode in (DIR_BOTH, DIR_LONG_ONLY) + + @property + def allows_short(self) -> bool: + if not self.direction_restrict_enabled: + return True + return self.direction_mode in (DIR_BOTH, DIR_SHORT_ONLY) + + +def load_trade_policy(env: Optional[dict] = None) -> TradePolicy: + e = env if env is not None else os.environ + direction_restrict = _env_bool(e.get("TRADE_DIRECTION_RESTRICT_ENABLED"), False) + symbol_restrict = _env_bool(e.get("TRADE_SYMBOL_RESTRICT_ENABLED"), False) + direction_mode = normalize_direction_mode(e.get("TRADE_DIRECTION")) + whitelist = parse_symbol_whitelist(e.get("TRADE_SYMBOL_WHITELIST")) + if symbol_restrict and not whitelist: + symbol_restrict = False + return TradePolicy( + direction_restrict_enabled=direction_restrict, + direction_mode=direction_mode, + symbol_restrict_enabled=symbol_restrict, + symbol_whitelist=whitelist, + ) + + +def direction_mode_label_zh(mode: str) -> str: + m = normalize_direction_mode(mode) + if m == DIR_LONG_ONLY: + return "仅多" + if m == DIR_SHORT_ONLY: + return "仅空" + return "双向" + + +def trade_policy_badge_parts(policy: TradePolicy) -> Tuple[str, ...]: + parts: list[str] = [] + if policy.direction_restrict_enabled: + if policy.direction_mode == DIR_LONG_ONLY: + parts.append("仅多") + elif policy.direction_mode == DIR_SHORT_ONLY: + parts.append("仅空") + if policy.symbol_restrict_enabled and policy.symbol_whitelist: + parts.append("/".join(policy.symbol_whitelist)) + return tuple(parts) + + +def trade_policy_to_dict(policy: TradePolicy) -> dict: + badges = trade_policy_badge_parts(policy) + return { + "direction_restrict_enabled": policy.direction_restrict_enabled, + "direction_mode": policy.direction_mode, + "direction_label_zh": ( + direction_mode_label_zh(policy.direction_mode) + if policy.direction_restrict_enabled + else "双向" + ), + "allows_long": policy.allows_long, + "allows_short": policy.allows_short, + "symbol_restrict_enabled": policy.symbol_restrict_enabled, + "symbol_whitelist": list(policy.symbol_whitelist), + "badge_parts": list(badges), + "badge_text": " · ".join(badges), + } + + +def normalize_open_direction(policy: TradePolicy, direction: str) -> str: + d = (direction or "long").strip().lower() + if d not in ("long", "short"): + d = "long" + if policy.direction_restrict_enabled: + if policy.direction_mode == DIR_LONG_ONLY: + return "long" + if policy.direction_mode == DIR_SHORT_ONLY: + return "short" + return d + + +def assert_direction_allowed(policy: TradePolicy, direction: str) -> Tuple[bool, str]: + d = (direction or "").strip().lower() + if d not in ("long", "short"): + if d in ("watch", ""): + return True, "" + return False, "方向无效，请选择做多或做空" + if d == "long" and not policy.allows_long: + return False, "当前账户配置为仅做空，不允许做多" + if d == "short" and not policy.allows_short: + return False, "当前账户配置为仅做多，不允许做空" + return True, "" + + +def assert_symbol_allowed( + policy: TradePolicy, + symbol: str, + *, + normalize_symbol_fn: Optional[Callable[[str], str]] = None, +) -> Tuple[bool, str]: + if not policy.symbol_restrict_enabled: + return True, "" + sym = (symbol or "").strip() + if not sym: + return False, "请选择币种" + if normalize_symbol_fn is not None: + sym_norm = (normalize_symbol_fn(sym) or "").strip() + else: + sym_norm = sym + base = symbol_base_coin(sym_norm or sym) + allowed: FrozenSet[str] = frozenset(policy.symbol_whitelist) + if base not in allowed: + allowed_txt = "、".join(policy.symbol_whitelist) + return False, f"当前账户仅允许 {allowed_txt}，不允许 {base or sym}" + return True, "" + + +def assert_trade_policy_open( + policy: TradePolicy, + symbol: str, + direction: str, + normalize_symbol_fn: Optional[Callable[[str], str]] = None, +) -> Tuple[bool, str]: + ok_sym, msg_sym = assert_symbol_allowed( + policy, symbol, normalize_symbol_fn=normalize_symbol_fn + ) + if not ok_sym: + return False, msg_sym + ok_dir, msg_dir = assert_direction_allowed(policy, direction) + if not ok_dir: + return False, msg_dir + return True, "" diff --git a/scripts/sync_trade_policy_env.py b/scripts/sync_trade_policy_env.py new file mode 100644 index 0000000..b267ac9 --- /dev/null +++ b/scripts/sync_trade_policy_env.py @@ -0,0 +1,216 @@ +#!/usr/bin/env python3 +""" +将账户方向 / 币种白名单 env 写入三所 .env（缺失则追加，已存在则 --set 时覆盖）。 + +用法（仓库根目录）： + python scripts/sync_trade_policy_env.py + python scripts/sync_trade_policy_env.py --dry-run + python scripts/sync_trade_policy_env.py --apply-account-profiles + python scripts/sync_trade_policy_env.py --set-direction binance long_only + +--apply-account-profiles：币安=仅多，Gate=BTC/ETH 白名单，OKX=默认不限制。 +修改后须 pm2 restart 对应实例。 +""" +from __future__ import annotations + +import argparse +import os +import re + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +INSTANCES = ( + "crypto_monitor_binance", + "crypto_monitor_okx", + "crypto_monitor_gate", +) + +COMMENT_BLOCK = [ + "# 方向限制（默认 false=双向均可；true 时按 TRADE_DIRECTION 限制，修改后须重启）", + "# TRADE_DIRECTION=long_only | short_only | both（或 多/空/双向）", + "# 币种白名单（默认 false=全币种可手输；true 时关键位/下单/策略仅下拉选择）", +] + +DEFAULTS = { + "TRADE_DIRECTION_RESTRICT_ENABLED": "false", + "TRADE_DIRECTION": "both", + "TRADE_SYMBOL_RESTRICT_ENABLED": "false", + "TRADE_SYMBOL_WHITELIST": "BTC,ETH", +} + +ACCOUNT_PROFILES = { + "crypto_monitor_binance": { + "TRADE_DIRECTION_RESTRICT_ENABLED": "true", + "TRADE_DIRECTION": "long_only", + "TRADE_SYMBOL_RESTRICT_ENABLED": "false", + "TRADE_SYMBOL_WHITELIST": "BTC,ETH", + }, + "crypto_monitor_gate": { + "TRADE_DIRECTION_RESTRICT_ENABLED": "false", + "TRADE_DIRECTION": "both", + "TRADE_SYMBOL_RESTRICT_ENABLED": "true", + "TRADE_SYMBOL_WHITELIST": "BTC,ETH", + }, + "crypto_monitor_okx": { + "TRADE_DIRECTION_RESTRICT_ENABLED": "false", + "TRADE_DIRECTION": "both", + "TRADE_SYMBOL_RESTRICT_ENABLED": "false", + "TRADE_SYMBOL_WHITELIST": "BTC,ETH", + }, +} + + +def _parse_env(path: str) -> list[str]: + if not os.path.isfile(path): + return [] + with open(path, "r", encoding="utf-8", errors="ignore") as f: + return f.read().replace("\r\n", "\n").replace("\r", "\n").splitlines() + + +def _env_get(lines: list[str], key: str) -> str | None: + pat = re.compile(r"^\s*" + re.escape(key) + r"\s*=\s*(.*)\s*$") + for line in lines: + m = pat.match(line) + if m: + return m.group(1).strip().strip('"').strip("'") + return None + + +def _upsert(lines: list[str], key: str, value: str) -> list[str]: + pat = re.compile(r"^\s*" + re.escape(key) + r"\s*=") + out: list[str] = [] + replaced = False + for line in lines: + if pat.match(line): + if not replaced: + out.append(f"{key}={value}") + replaced = True + continue + out.append(line) + if not replaced: + if out and out[-1].strip(): + out.append("") + out.append(f"{key}={value}") + return out + + +def _insert_after(lines: list[str], anchor_key: str, insert: list[str]) -> list[str]: + pat = re.compile(r"^\s*" + re.escape(anchor_key) + r"\s*=") + for i, line in enumerate(lines): + if pat.match(line): + return lines[: i + 1] + insert + lines[i + 1 :] + if lines and lines[-1].strip(): + return lines + [""] + insert + return lines + insert + + +def sync_one( + dir_name: str, + values: dict[str, str], + *, + dry_run: bool, + force: bool, +) -> bool: + path = os.path.join(REPO, dir_name, ".env") + if not os.path.isfile(path): + print(f"skip (no .env): {dir_name}") + return False + lines = _parse_env(path) + changed = False + for key, val in values.items(): + cur = _env_get(lines, key) + if cur is None: + if key == "TRADE_DIRECTION_RESTRICT_ENABLED" and _env_get( + lines, "TRADE_DIRECTION" + ) is None: + if COMMENT_BLOCK[0] not in "\n".join(lines): + lines = _insert_after(lines, "POSITION_SIZING_MODE", COMMENT_BLOCK) + lines = _upsert(lines, key, val) + changed = True + elif force or cur != val: + lines = _upsert(lines, key, val) + changed = True + if not changed: + print(f"ok (unchanged): {dir_name}") + return False + text = "\n".join(lines).rstrip() + "\n" + print(f"update: {dir_name}") + for k, v in values.items(): + print(f" {k}={v}") + if not dry_run: + with open(path, "w", encoding="utf-8", newline="\n") as f: + f.write(text) + return True + + +def main() -> None: + ap = argparse.ArgumentParser(description="同步三所 trade policy env") + ap.add_argument("--dry-run", action="store_true") + ap.add_argument( + "--apply-account-profiles", + action="store_true", + help="币安仅多、Gate BTC/ETH、OKX 默认", + ) + ap.add_argument( + "--defaults-only", + action="store_true", + help="三所均写入默认（不限制）", + ) + ap.add_argument("--force", action="store_true", help="覆盖已有值") + ap.add_argument("--set-direction", nargs=2, metavar=("INSTANCE", "MODE")) + ap.add_argument("--set-symbol-whitelist", nargs=2, metavar=("INSTANCE", "SYMS")) + args = ap.parse_args() + + if args.set_direction: + inst, mode = args.set_direction + if inst not in INSTANCES: + raise SystemExit(f"unknown instance: {inst}") + sync_one( + inst, + { + "TRADE_DIRECTION_RESTRICT_ENABLED": "true", + "TRADE_DIRECTION": mode, + }, + dry_run=args.dry_run, + force=True, + ) + return + + if args.set_symbol_whitelist: + inst, syms = args.set_symbol_whitelist + if inst not in INSTANCES: + raise SystemExit(f"unknown instance: {inst}") + sync_one( + inst, + { + "TRADE_SYMBOL_RESTRICT_ENABLED": "true", + "TRADE_SYMBOL_WHITELIST": syms, + }, + dry_run=args.dry_run, + force=True, + ) + return + + profiles = ( + {k: dict(DEFAULTS) for k in INSTANCES} + if args.defaults_only + else dict(ACCOUNT_PROFILES) + if args.apply_account_profiles + else {k: dict(DEFAULTS) for k in INSTANCES} + ) + + if not args.apply_account_profiles and not args.defaults_only: + ap.print_help() + print("\n提示：部署常用 --apply-account-profiles") + return + + any_changed = False + for inst in INSTANCES: + if sync_one(inst, profiles[inst], dry_run=args.dry_run, force=args.force): + any_changed = True + if args.dry_run and any_changed: + print("(dry-run, 未写入)") + + +if __name__ == "__main__": + main() diff --git a/tests/test_trade_policy_lib.py b/tests/test_trade_policy_lib.py new file mode 100644 index 0000000..efa32fa --- /dev/null +++ b/tests/test_trade_policy_lib.py @@ -0,0 +1,90 @@ +"""账户方向 / 币种白名单 env 策略。""" +from lib.trade.trade_policy_lib import ( + assert_direction_allowed, + assert_symbol_allowed, + assert_trade_policy_open, + load_trade_policy, + parse_symbol_whitelist, + symbol_base_coin, + trade_policy_badge_parts, +) + + +def test_default_policy_unrestricted(): + p = load_trade_policy({}) + assert not p.direction_restrict_enabled + assert not p.symbol_restrict_enabled + assert p.allows_long and p.allows_short + + +def test_long_only_blocks_short(): + p = load_trade_policy( + { + "TRADE_DIRECTION_RESTRICT_ENABLED": "true", + "TRADE_DIRECTION": "long_only", + } + ) + ok, msg = assert_direction_allowed(p, "short") + assert not ok + assert "仅做多" in msg + ok2, _ = assert_direction_allowed(p, "long") + assert ok2 + + +def test_symbol_whitelist_btc_eth(): + p = load_trade_policy( + { + "TRADE_SYMBOL_RESTRICT_ENABLED": "true", + "TRADE_SYMBOL_WHITELIST": "BTC,ETH", + } + ) + ok, _ = assert_symbol_allowed(p, "BTC/USDT") + assert ok + ok2, msg = assert_symbol_allowed(p, "SOL") + assert not ok2 + assert "SOL" in msg + + +def test_symbol_whitelist_without_list_disables_restrict(): + p = load_trade_policy( + { + "TRADE_SYMBOL_RESTRICT_ENABLED": "true", + "TRADE_SYMBOL_WHITELIST": "", + } + ) + assert not p.symbol_restrict_enabled + + +def test_combined_open_validation(): + p = load_trade_policy( + { + "TRADE_DIRECTION_RESTRICT_ENABLED": "1", + "TRADE_DIRECTION": "多", + "TRADE_SYMBOL_RESTRICT_ENABLED": "yes", + "TRADE_SYMBOL_WHITELIST": "BTC,ETH", + } + ) + ok, _ = assert_trade_policy_open(p, "ETH", "long") + assert ok + ok2, msg = assert_trade_policy_open(p, "ETH", "short") + assert not ok2 + ok3, msg3 = assert_trade_policy_open(p, "BNB", "long") + assert not ok3 + assert "BNB" in msg3 + + +def test_parse_whitelist_and_base_coin(): + assert parse_symbol_whitelist("btc, eth") == ("BTC", "ETH") + assert symbol_base_coin("btc/usdt:usdt") == "BTC" + + +def test_badge_parts(): + p = load_trade_policy( + { + "TRADE_DIRECTION_RESTRICT_ENABLED": "true", + "TRADE_DIRECTION": "long_only", + "TRADE_SYMBOL_RESTRICT_ENABLED": "true", + "TRADE_SYMBOL_WHITELIST": "BTC,ETH", + } + ) + assert trade_policy_badge_parts(p) == ("仅多", "BTC/ETH")