dekun/crypto_monitor
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: add unified four-exchange env sync scripts

Browse Source 
sync_four_exchange_env runs position sizing + transfer sync; transfer script preserves existing values and supports --set-amount/--enable-auto-transfer.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-06-04 09:59:33 +08:00
parent 29b0634c6d
commit e6e79215fc
4 changed files with 188 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/sync_four_exchange_env.py
+58
View File
@@ -0,0 +1,58 @@
#!/usr/bin/env python3
"""
四所 .env 一次性同步：计仓模式 + 自动划转（调用子脚本，不覆盖已有自定义值）。
用法（仓库根目录）：
python scripts/sync_four_exchange_env.py
python scripts/sync_four_exchange_env.py --dry-run
python scripts/sync_four_exchange_env.py --set-transfer-amount 50 --enable-auto-transfer
子脚本可单独运行：
python scripts/sync_four_exchange_position_sizing_env.py
python scripts/sync_four_exchange_transfer_env.py
"""
from __future__ import annotations
import argparse
import subprocess
import sys
from pathlib import Path
REPO = Path(__file__).resolve().parent.parent
PY = sys.executable
def _run(script: str, extra: list[str]) -> int:
cmd = [PY, str(REPO / "scripts" / script)] + extra
print(f"\n>>> {' '.join(cmd)}")
return subprocess.call(cmd, cwd=str(REPO))
def main():
ap = argparse.ArgumentParser(description="四所 .env 统一同步（计仓 + 划转）")
ap.add_argument("--dry-run", action="store_true")
ap.add_argument("--set-mode", choices=("risk", "full_margin"), metavar="MODE")
ap.add_argument("--set-transfer-amount", metavar="U")
ap.add_argument("--enable-auto-transfer", action="store_true")
args = ap.parse_args()
dry = ["--dry-run"] if args.dry_run else []
code = 0
ps_args = list(dry)
if args.set_mode:
ps_args.extend(["--set-mode", args.set_mode])
code |= _run("sync_four_exchange_position_sizing_env.py", ps_args)
tr_args = list(dry)
if args.set_transfer_amount:
tr_args.extend(["--set-amount", args.set_transfer_amount])
if args.enable_auto_transfer:
tr_args.append("--enable-auto-transfer")
code |= _run("sync_four_exchange_transfer_env.py", tr_args)
sys.exit(code)
if __name__ == "__main__":
main()
scripts/sync_four_exchange_transfer_env.py
+118 -22
View File
@@ -1,10 +1,12 @@
#!/usr/bin/env python3
"""
将自动划转 / 币安资金账户相关项写入四所实例 .env（已存在则更新，缺失则追加）。
将每日自动划转相关项写入四所实例 .env（已有值保留，缺失则追加；可选强制改金额/开关）。
用法（仓库根目录）：
python scripts/sync_four_exchange_transfer_env.py
python scripts/sync_four_exchange_transfer_env.py --dry-run
python scripts/sync_four_exchange_transfer_env.py --set-amount 50
python scripts/sync_four_exchange_transfer_env.py --enable-auto-transfer
不修改 API 密钥与其它自定义项；若 .env 不存在则跳过（请先从 .env.example 复制）。
"""
@@ -23,8 +25,12 @@ INSTANCES = (
"crypto_monitor_gate_bot",
)
# 四所统一（页顶「将 swap 调整至 XU」= AUTO_TRANSFER_AMOUNT，双向归集，与 DAILY_START_CAPITAL 独立）
COMMON_KEYS = {
COMMENT_BLOCK = (
"# 自动划转：北京时间 AUTO_TRANSFER_BJ_HOUR 点将 swap 调整至 AUTO_TRANSFER_AMOUNT；"
"不足 funding→swap、超出 swap→funding；持仓中不划转"
)
DEFAULTS = {
"AUTO_TRANSFER_ENABLED": "false",
"AUTO_TRANSFER_FROM": "funding",
"AUTO_TRANSFER_TO": "swap",
@@ -32,6 +38,8 @@ COMMON_KEYS = {
"AUTO_TRANSFER_BJ_HOUR": "8",
}
DEFAULT_AMOUNT = "50"
BINANCE_ONLY = {
"BINANCE_FUNDING_INCLUDE_SPOT": "false",
}
@@ -71,44 +79,132 @@ def _upsert(lines: list[str], key: str, value: str) -> list[str]:
return out
def _ensure_transfer_block(lines: list[str], extra: dict[str, str]) -> list[str]:
daily = _env_get(lines, "DAILY_START_CAPITAL")
def _insert_before(lines: list[str], anchor_key: str, insert: list[str]) -> list[str]:
pat = re.compile(r"^\s*" + re.escape(anchor_key) + r"\s*=")
for i, line in enumerate(lines):
if pat.match(line):
return lines[:i] + insert + lines[i:]
if lines and lines[-1].strip():
return lines + [""] + insert
return lines + insert
def _resolve_default_amount(lines: list[str]) -> str:
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)
if amount is not None:
return amount
daily = _env_get(lines, "DAILY_START_CAPITAL")
if daily is not None:
return daily
return DEFAULT_AMOUNT
def _ensure_key(
lines: list[str],
key: str,
value: str,
*,
force: bool,
) -> list[str]:
if force or _env_get(lines, key) is None:
return _upsert(lines, key, value)
return lines
def sync_one(dir_name: str, dry_run: bool) -> str:
def _ensure_transfer_block(
lines: list[str],
extra: dict[str, str],
*,
force_amount: str | None,
force_enabled: str | None,
) -> list[str]:
amount = force_amount if force_amount is not None else _resolve_default_amount(lines)
had_amount = _env_get(lines, "AUTO_TRANSFER_AMOUNT") is not None
if not had_amount and COMMENT_BLOCK not in lines:
lines = _insert_before(
lines,
"AUTO_TRANSFER_ENABLED",
[COMMENT_BLOCK],
)
if _env_get(lines, "AUTO_TRANSFER_ENABLED") is None:
lines = _insert_before(
lines,
"BALANCE_REFRESH_SECONDS",
[COMMENT_BLOCK],
)
lines = _ensure_key(
lines,
"AUTO_TRANSFER_AMOUNT",
amount,
force=force_amount is not None,
)
for k, v in DEFAULTS.items():
if k == "AUTO_TRANSFER_ENABLED" and force_enabled is not None:
lines = _upsert(lines, k, force_enabled)
else:
lines = _ensure_key(lines, k, v, force=False)
for k, v in extra.items():
lines = _ensure_key(lines, k, v, force=False)
return lines
def sync_one(
dir_name: str,
dry_run: bool,
*,
set_amount: str | None,
enable_auto: bool | None,
) -> 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)
old_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}: 已是最新"
force_enabled = "true" if enable_auto is True else None
new_lines = _ensure_transfer_block(
old_lines,
extra,
force_amount=set_amount,
force_enabled=force_enabled,
)
enabled = _env_get(new_lines, "AUTO_TRANSFER_ENABLED") or DEFAULTS["AUTO_TRANSFER_ENABLED"]
amt = _env_get(new_lines, "AUTO_TRANSFER_AMOUNT") or DEFAULT_AMOUNT
if new_lines == old_lines:
return f"OK {dir_name}: ENABLED={enabled} AMOUNT={amt}"
if dry_run:
return f"DRY {dir_name}: 将更新 {len([1 for a,b in zip(lines,new_lines) if a!=b])}+ 行"
return f"DRY {dir_name}: 将更新 ENABLED={enabled} AMOUNT={amt}"
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}"
return f"DONE {dir_name}: ENABLED={enabled} AMOUNT={amt}"
def main():
ap = argparse.ArgumentParser()
ap = argparse.ArgumentParser(description="四所 .env 自动划转项同步")
ap.add_argument("--dry-run", action="store_true")
ap.add_argument(
"--set-amount",
metavar="U",
help=f"强制四所 AUTO_TRANSFER_AMOUNT（缺省补全默认 {DEFAULT_AMOUNT}）",
)
ap.add_argument(
"--enable-auto-transfer",
action="store_true",
help="强制四所 AUTO_TRANSFER_ENABLED=true",
)
args = ap.parse_args()
for name in INSTANCES:
print(sync_one(name, args.dry_run))
print(
sync_one(
name,
args.dry_run,
set_amount=args.set_amount,
enable_auto=True if args.enable_auto_transfer else None,
)
)
if __name__ == "__main__":