Files
eth_hedge_sim/backend/app/env_store.py
T
dekun dbc86a1ce6 Add Binance live trading, anti-stuck open/close recovery, and configurable rate limits.
OKX/Binance LIVE share half_open and option_closed_perp_pending repair paths; private REST throttles default to 1s and are tunable in settings.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-26 21:35:10 +08:00

68 lines
2.1 KiB
Python

"""批量写入 .env 并刷新 Settings 缓存。"""
from __future__ import annotations
from pathlib import Path
from .config import get_settings
from .credentials import upsert_env_file
def upsert_env_keys(updates: dict[str, str]) -> Path | None:
"""写入多项;空 value 跳过。返回最后写入的 .env 路径。"""
target: Path | None = None
for key, value in updates.items():
if value is None:
continue
# 允许显式清空密钥(传空串以外的 sentinel 由调用方决定);空串表示跳过
if value == "":
continue
target = upsert_env_file(key, value)
get_settings.cache_clear()
return target
def mask_secret(raw: str | None, *, keep: int = 4) -> str | None:
"""脱敏:****末尾;过短则全部打码。"""
s = (raw or "").strip()
if not s:
return None
if len(s) <= keep:
return "*" * len(s)
return "*" * max(4, len(s) - keep) + s[-keep:]
def okx_keys_configured(s=None) -> bool:
st = s or get_settings()
return bool(
(st.okx_api_key or "").strip()
and (st.okx_api_secret or "").strip()
and (st.okx_api_passphrase or "").strip()
)
def binance_keys_configured(s=None) -> bool:
st = s or get_settings()
return bool(
(st.binance_api_key or "").strip() and (st.binance_api_secret or "").strip()
)
def live_ready(*, exchange: str | None = None) -> tuple[bool, str]:
"""LIVE 是否可下单。返回 (ok, reason)。"""
from .exchange.runtime import load_runtime_settings, normalize_exchange_name
st = get_settings()
if st.is_sim:
return True, "sim"
ex = normalize_exchange_name(exchange or load_runtime_settings().exchange)
if ex == "binance":
if not binance_keys_configured(st):
return False, "币安 API Key/Secret 未配置"
return True, "ok"
if ex == "okx":
if not okx_keys_configured(st):
return False, "OKX API Key/Secret/Passphrase 未配置"
return True, "ok"
return False, f"未知交易所: {ex}"