dekun
58 lines
1.8 KiB
Python
58 lines
1.8 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 "").strip()
|
|
if policy.symbol_restrict_enabled and policy.symbol_whitelist:
|
|
# 白名单仅一币时直接用 env 币种,表单下拉同步默认选中
|
|
if len(policy.symbol_whitelist) == 1:
|
|
return f"{policy.symbol_whitelist[0]}/USDT"
|
|
from lib.trade.trade_policy_lib import symbol_base_coin
|
|
|
|
base = symbol_base_coin(d or "BTC/USDT")
|
|
if base not in policy.symbol_whitelist:
|
|
return f"{policy.symbol_whitelist[0]}/USDT"
|
|
if d:
|
|
return d if "/" in d else f"{base}/USDT"
|
|
return f"{policy.symbol_whitelist[0]}/USDT"
|
|
return d or "BTC/USDT"
|
|
|
|
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
|
|
)
|