dekun
c0ad50a7b5
Per-instance TRADE_DIRECTION / TRADE_SYMBOL_WHITELIST restricts UI and API for manual orders, key monitors, and strategies; includes sync script for deployment profiles. Co-authored-by: Cursor <cursoragent@cursor.com>
53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
"""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
|
|
)
|