dekun
95 lines
2.5 KiB
Python
95 lines
2.5 KiB
Python
"""交易所 API 凭证规范化.
|
|
|
|
新机 .env 密钥应为空;示例占位符不得注入 ccxt,否则鉴权失败且
|
|
(尤其 Gate)反复签名请求易触发 IP 封禁.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
|
|
_PLACEHOLDER_EXACT = frozenset(
|
|
{
|
|
"你的密钥",
|
|
"your-api-key",
|
|
"your_api_key",
|
|
"your-api-secret",
|
|
"your_api_secret",
|
|
"todo",
|
|
"xxx",
|
|
"changeme",
|
|
}
|
|
)
|
|
|
|
|
|
def normalize_api_credential(value: Optional[str]) -> str:
|
|
"""去空白;空/占位符一律视为未配置."""
|
|
s = (value or "").strip().strip('"').strip("'")
|
|
if not s:
|
|
return ""
|
|
upper = s.upper()
|
|
if upper.startswith("REPLACE_WITH"):
|
|
return ""
|
|
if upper.startswith("CHANGE_TO"):
|
|
return ""
|
|
if s.lower() in _PLACEHOLDER_EXACT:
|
|
return ""
|
|
return s
|
|
|
|
|
|
def credentials_configured(*parts: Optional[str]) -> bool:
|
|
return all(bool(normalize_api_credential(p)) for p in parts)
|
|
|
|
|
|
def is_exchange_auth_error(exc: BaseException) -> bool:
|
|
"""鉴权/无效 Key 类错误(用于停掉后续签名请求,避免 Gate 封 IP)."""
|
|
name = type(exc).__name__
|
|
if name in ("AuthenticationError", "PermissionDenied", "InvalidNonce"):
|
|
return True
|
|
msg = str(exc)
|
|
markers = (
|
|
"Invalid Api-Key",
|
|
"Invalid API-key",
|
|
"Invalid API Key",
|
|
"INVALID_KEY",
|
|
"Invalid key",
|
|
"API key is invalid",
|
|
"api key not found",
|
|
"Signature",
|
|
"INVALID_SIGNATURE",
|
|
"401",
|
|
"-2008",
|
|
"-2014",
|
|
"-2015",
|
|
"10003", # Gate: invalid key often
|
|
"INVALID_KEY",
|
|
)
|
|
low = msg.lower()
|
|
if "api" in low and ("key" in low or "sign" in low) and (
|
|
"invalid" in low or "incorrect" in low or "not found" in low
|
|
):
|
|
return True
|
|
return any(m in msg for m in markers)
|
|
|
|
|
|
def strip_ccxt_credentials(exchange: Any) -> None:
|
|
"""内存中清空密钥,后续只走公开接口,避免继续带坏钥签名."""
|
|
try:
|
|
exchange.apiKey = ""
|
|
except Exception:
|
|
pass
|
|
try:
|
|
exchange.secret = ""
|
|
except Exception:
|
|
pass
|
|
try:
|
|
exchange.password = ""
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def load_markets_public_fallback(exchange: Any, *, reload: bool = False) -> None:
|
|
"""鉴权失败后去掉密钥再拉公开 markets(最多再请求一次)."""
|
|
strip_ccxt_credentials(exchange)
|
|
exchange.load_markets(reload=reload)
|