docs(env): unify auto-transfer config across four exchanges

Align .env.example blocks and add sync script to merge AUTO_TRANSFER_* into existing instance .env files without touching API secrets.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-06-04 07:47:18 +08:00
parent b1952dcd6e
commit d75527a9ca
6 changed files with 149 additions and 8 deletions
+115
View File
@@ -0,0 +1,115 @@
#!/usr/bin/env python3
"""
将自动划转 / 币安资金账户相关项写入四所实例 .env(已存在则更新,缺失则追加)。
用法(仓库根目录):
python scripts/sync_four_exchange_transfer_env.py
python scripts/sync_four_exchange_transfer_env.py --dry-run
不修改 API 密钥与其它自定义项;若 .env 不存在则跳过(请先从 .env.example 复制)。
"""
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",
"crypto_monitor_gate_bot",
)
# 四所统一(页顶「将 swap 补足到 XU」= AUTO_TRANSFER_AMOUNT,与 DAILY_START_CAPITAL 独立)
COMMON_KEYS = {
"AUTO_TRANSFER_ENABLED": "false",
"AUTO_TRANSFER_FROM": "funding",
"AUTO_TRANSFER_TO": "swap",
"TRANSFER_CCY": "USDT",
"AUTO_TRANSFER_BJ_HOUR": "8",
}
BINANCE_ONLY = {
"BINANCE_FUNDING_INCLUDE_SPOT": "false",
}
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 = []
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 _ensure_transfer_block(lines: list[str], extra: dict[str, str]) -> list[str]:
daily = _env_get(lines, "DAILY_START_CAPITAL")
amount = _env_get(lines, "AUTO_TRANSFER_AMOUNT")
if amount is None:
amount = daily if daily else "30"
keys = dict(COMMON_KEYS)
keys["AUTO_TRANSFER_AMOUNT"] = amount
keys.update(extra)
for k, v in keys.items():
lines = _upsert(lines, k, v)
return lines
def sync_one(dir_name: str, dry_run: bool) -> str:
env_path = os.path.join(REPO, dir_name, ".env")
if not os.path.isfile(env_path):
return f"SKIP {dir_name}: 无 .env(请 cp .env.example .env"
lines = _parse_env(env_path)
extra = dict(BINANCE_ONLY) if dir_name == "crypto_monitor_binance" else {}
new_lines = _ensure_transfer_block(lines, extra)
if new_lines == lines:
return f"OK {dir_name}: 已是最新"
if dry_run:
return f"DRY {dir_name}: 将更新 {len([1 for a,b in zip(lines,new_lines) if a!=b])}+ 行"
with open(env_path, "w", encoding="utf-8", newline="\n") as f:
f.write("\n".join(new_lines))
if new_lines and new_lines[-1].strip():
f.write("\n")
amt = _env_get(new_lines, "AUTO_TRANSFER_AMOUNT")
return f"DONE {dir_name}: AUTO_TRANSFER_AMOUNT={amt}"
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--dry-run", action="store_true")
args = ap.parse_args()
for name in INSTANCES:
print(sync_one(name, args.dry_run))
if __name__ == "__main__":
main()