refactor: 移除 gate_bot，统一为三所架构并更新文档
删除 crypto_monitor_gate_bot 目录，中控与子代理改为 binance/okx/gate 三账户； 文档与 UI 文案「四所」改为「三所」；新增清库前一次性配置备份脚本。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
@@ -1,5 +1,5 @@
|
||||
#!/usr/bin/env python3
|
||||
"""对 binance/okx/gate_bot 应用与 gate 相同的时间平仓代码替换。"""
|
||||
"""对 binance/okx 应用与 gate 相同的时间平仓代码替换。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
@@ -8,7 +8,6 @@ ROOT = Path(__file__).resolve().parents[1]
|
||||
FILES = [
|
||||
ROOT / "crypto_monitor_binance" / "app.py",
|
||||
ROOT / "crypto_monitor_okx" / "app.py",
|
||||
ROOT / "crypto_monitor_gate_bot" / "app.py",
|
||||
]
|
||||
|
||||
REPLACEMENTS: list[tuple[str, str]] = [
|
||||
|
||||
@@ -1,14 +1,14 @@
|
||||
#!/usr/bin/env python3
|
||||
"""补录缺失的趋势回调策略结束快照（strategy_trade_snapshots）。
|
||||
|
||||
适用：gate_bot 等在计划结束（止盈/止损/手动）时因 strategy_trend_cfg 未注册而漏写快照的历史数据。
|
||||
适用：gate 等在计划结束（止盈/止损/手动）时因 strategy_trend_cfg 未注册而漏写快照的历史数据。
|
||||
保本移交路径通常已有快照，本脚本默认跳过「已有任意快照」的计划。
|
||||
|
||||
用法（在仓库根目录，Linux 请用 python3）：
|
||||
python3 scripts/backfill_trend_strategy_snapshots.py \\
|
||||
--db crypto_monitor_gate_bot/crypto.db --dry-run
|
||||
--db crypto_monitor_gate/crypto.db --dry-run
|
||||
python3 scripts/backfill_trend_strategy_snapshots.py \\
|
||||
--db crypto_monitor_gate_bot/crypto.db --apply
|
||||
--db crypto_monitor_gate/crypto.db --apply
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
#!/usr/bin/env python3
|
||||
"""补录缺失的趋势回调 trade_records（策略快照已有、交易记录漏写）。
|
||||
|
||||
典型原因：gate_bot insert_trade_record 曾不接受 entry_reason，_finalize_plan 写快照后插入失败。
|
||||
典型原因：gate insert_trade_record 曾不接受 entry_reason，_finalize_plan 写快照后插入失败。
|
||||
|
||||
用法：
|
||||
python scripts/backfill_trend_trade_records.py --db crypto_monitor_gate_bot/crypto.db --dry-run
|
||||
python scripts/backfill_trend_trade_records.py --db crypto_monitor_gate_bot/crypto.db --apply
|
||||
python scripts/backfill_trend_trade_records.py --db crypto_monitor_gate/crypto.db --dry-run
|
||||
python scripts/backfill_trend_trade_records.py --db crypto_monitor_gate/crypto.db --apply
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
#!/usr/bin/env python3
|
||||
"""清理 strategy_trade_snapshots 重复行（同计划 + 同结果仅保留 id 最大的一条）。
|
||||
|
||||
用法（在实例目录，如 crypto_monitor_gate_bot）：
|
||||
用法（在实例目录，如 crypto_monitor_gate）：
|
||||
python ../scripts/dedupe_strategy_snapshots.py
|
||||
python ../scripts/dedupe_strategy_snapshots.py --db crypto.db
|
||||
"""
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
一次性备份：三所 .env + 中控 .env / hub_settings.json（不含图片、不含数据库）。
|
||||
|
||||
用途：删除 gate、清库、全新计划启动前，在仓库根目录执行一次即可：
|
||||
|
||||
python scripts/one_shot_backup_config_before_cleanup.py
|
||||
|
||||
输出目录默认：backups/one-shot-YYYYMMDD-HHMMSS/config/
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
CONFIG_SOURCES: list[tuple[str, Path]] = [
|
||||
("crypto_monitor_binance.env", REPO_ROOT / "crypto_monitor_binance" / ".env"),
|
||||
("crypto_monitor_okx.env", REPO_ROOT / "crypto_monitor_okx" / ".env"),
|
||||
("crypto_monitor_gate.env", REPO_ROOT / "crypto_monitor_gate" / ".env"),
|
||||
("manual_trading_hub.env", REPO_ROOT / "manual_trading_hub" / ".env"),
|
||||
("hub_settings.json", REPO_ROOT / "manual_trading_hub" / "hub_settings.json"),
|
||||
]
|
||||
|
||||
ENV_BACKUP_GLOBS = (
|
||||
REPO_ROOT / "crypto_monitor_binance",
|
||||
REPO_ROOT / "crypto_monitor_okx",
|
||||
REPO_ROOT / "crypto_monitor_gate",
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
|
||||
out_dir = REPO_ROOT / "backups" / f"one-shot-{stamp}" / "config"
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
copied: list[str] = []
|
||||
missing: list[str] = []
|
||||
|
||||
for dest_name, src in CONFIG_SOURCES:
|
||||
if src.is_file():
|
||||
shutil.copy2(src, out_dir / dest_name)
|
||||
copied.append(dest_name)
|
||||
else:
|
||||
missing.append(str(src.relative_to(REPO_ROOT)))
|
||||
|
||||
for inst_dir in ENV_BACKUP_GLOBS:
|
||||
for src in sorted(inst_dir.glob(".env.backup.*")):
|
||||
dest_name = f"{inst_dir.name}.{src.name}"
|
||||
shutil.copy2(src, out_dir / dest_name)
|
||||
copied.append(dest_name)
|
||||
|
||||
manifest = out_dir.parent / "manifest.txt"
|
||||
lines = [
|
||||
f"created_at={stamp}",
|
||||
f"repo={REPO_ROOT}",
|
||||
"",
|
||||
"copied:",
|
||||
*[f" - {name}" for name in copied],
|
||||
"",
|
||||
"missing (skipped):",
|
||||
*[f" - {p}" for p in missing],
|
||||
"",
|
||||
"not included: crypto.db, hub *.db, static/images, gate",
|
||||
]
|
||||
manifest.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
|
||||
print(f"Backup written to: {out_dir}")
|
||||
if copied:
|
||||
print("Copied:", ", ".join(copied))
|
||||
if missing:
|
||||
print("Missing (ok if fresh install):", ", ".join(missing))
|
||||
print(f"Manifest: {manifest}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -5,7 +5,7 @@ from __future__ import annotations
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
EXCHANGES = ("crypto_monitor_binance", "crypto_monitor_okx", "crypto_monitor_gate", "crypto_monitor_gate_bot")
|
||||
EXCHANGES = ("crypto_monitor_binance", "crypto_monitor_okx", "crypto_monitor_gate")
|
||||
FILES = ("index.html", "login.html", "key_focus_v2.html", "order_focus_v2.html")
|
||||
|
||||
SCRIPT_TAG = ' <script src="/static/instance_theme.js?v=4"></script>\n'
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
#!/usr/bin/env python3
|
||||
"""一次性：为 okx/gate/gate_bot 注入与 binance 一致的计仓模式补丁（已 patch 过则跳过）。"""
|
||||
"""一次性：为 okx/gate 注入与 binance 一致的计仓模式补丁（已 patch 过则跳过）。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
@@ -82,7 +82,6 @@ TEMPLATE_RULE = ''' <div class="rule-tip">
|
||||
APPS = [
|
||||
("crypto_monitor_okx", 4, "_market_open_for_key_monitor", True),
|
||||
("crypto_monitor_gate", 2, "_market_open_for_key_monitor", True),
|
||||
("crypto_monitor_gate_bot", 4, None, False),
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -19,7 +19,6 @@ EXCHANGE_DIRS = (
|
||||
"crypto_monitor_binance",
|
||||
"crypto_monitor_okx",
|
||||
"crypto_monitor_gate",
|
||||
"crypto_monitor_gate_bot",
|
||||
)
|
||||
|
||||
FILES = (
|
||||
|
||||
@@ -1,60 +1,60 @@
|
||||
#!/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
|
||||
|
||||
完整说明见 docs/env-sync-scripts.md
|
||||
"""
|
||||
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()
|
||||
#!/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
|
||||
|
||||
完整说明见 docs/env-sync-scripts.md
|
||||
"""
|
||||
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()
|
||||
|
||||
@@ -1,180 +1,179 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
将计仓模式相关项写入四所实例 .env（已存在则保留原值，缺失则追加默认值）。
|
||||
|
||||
用法（仓库根目录）：
|
||||
python scripts/sync_four_exchange_position_sizing_env.py
|
||||
python scripts/sync_four_exchange_position_sizing_env.py --dry-run
|
||||
python scripts/sync_four_exchange_position_sizing_env.py --set-mode risk
|
||||
python scripts/sync_four_exchange_position_sizing_env.py --set-mode full_margin
|
||||
|
||||
切换 POSITION_SIZING_MODE 须在交易所无持仓后执行，并 pm2 restart 对应实例。
|
||||
不修改 API 密钥与其它自定义项；若 .env 不存在则跳过（请先从 .env.example 复制）。
|
||||
|
||||
完整说明见 docs/env-sync-scripts.md
|
||||
"""
|
||||
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",
|
||||
)
|
||||
|
||||
COMMENT_POSITION_SIZING = (
|
||||
"# 计仓：risk=以损定仓（默认）；full_margin=合约可用×FULL_MARGIN_BUFFER_RATIO 全仓杠杆（须无仓后重启）"
|
||||
)
|
||||
COMMENT_BUFFER = "# 使用可用资金时的缓冲比例（如0.98代表用98%）"
|
||||
|
||||
DEFAULT_MODE = "risk"
|
||||
DEFAULT_BUFFER = "0.98"
|
||||
VALID_MODES = frozenset({"risk", "full_margin"})
|
||||
|
||||
|
||||
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 _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 _ensure_position_sizing(lines: list[str], *, force_mode: str | None) -> list[str]:
|
||||
if force_mode is not None:
|
||||
if COMMENT_POSITION_SIZING not in lines and not _env_get(lines, "POSITION_SIZING_MODE"):
|
||||
lines = _insert_before(lines, "DAILY_START_CAPITAL", [COMMENT_POSITION_SIZING])
|
||||
return _upsert(lines, "POSITION_SIZING_MODE", force_mode)
|
||||
|
||||
cur = _env_get(lines, "POSITION_SIZING_MODE")
|
||||
if cur is not None:
|
||||
norm = cur.strip().lower()
|
||||
if norm in VALID_MODES and norm != cur:
|
||||
return _upsert(lines, "POSITION_SIZING_MODE", norm)
|
||||
if norm not in VALID_MODES:
|
||||
return _upsert(lines, "POSITION_SIZING_MODE", DEFAULT_MODE)
|
||||
return lines
|
||||
|
||||
block = [COMMENT_POSITION_SIZING, f"POSITION_SIZING_MODE={DEFAULT_MODE}"]
|
||||
return _insert_before(lines, "DAILY_START_CAPITAL", block)
|
||||
|
||||
|
||||
def _ensure_buffer_ratio(lines: list[str], *, force_buffer: str | None) -> list[str]:
|
||||
if force_buffer is not None:
|
||||
if COMMENT_BUFFER not in lines and _env_get(lines, "FULL_MARGIN_BUFFER_RATIO") is None:
|
||||
lines = _insert_before(lines, "BALANCE_REFRESH_SECONDS", [COMMENT_BUFFER])
|
||||
return _upsert(lines, "FULL_MARGIN_BUFFER_RATIO", force_buffer)
|
||||
|
||||
if _env_get(lines, "FULL_MARGIN_BUFFER_RATIO") is not None:
|
||||
return lines
|
||||
|
||||
block = [COMMENT_BUFFER, f"FULL_MARGIN_BUFFER_RATIO={DEFAULT_BUFFER}"]
|
||||
return _insert_before(lines, "BALANCE_REFRESH_SECONDS", block)
|
||||
|
||||
|
||||
def sync_one(
|
||||
dir_name: str,
|
||||
dry_run: bool,
|
||||
*,
|
||||
set_mode: str | None,
|
||||
set_buffer: str | 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）"
|
||||
old_lines = _parse_env(env_path)
|
||||
new_lines = _ensure_buffer_ratio(
|
||||
_ensure_position_sizing(list(old_lines), force_mode=set_mode),
|
||||
force_buffer=set_buffer,
|
||||
)
|
||||
mode = _env_get(new_lines, "POSITION_SIZING_MODE") or DEFAULT_MODE
|
||||
buf = _env_get(new_lines, "FULL_MARGIN_BUFFER_RATIO") or DEFAULT_BUFFER
|
||||
if new_lines == old_lines:
|
||||
return f"OK {dir_name}: POSITION_SIZING_MODE={mode} FULL_MARGIN_BUFFER_RATIO={buf}"
|
||||
if dry_run:
|
||||
return (
|
||||
f"DRY {dir_name}: 将写入 POSITION_SIZING_MODE={mode} "
|
||||
f"FULL_MARGIN_BUFFER_RATIO={buf}"
|
||||
)
|
||||
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")
|
||||
return f"DONE {dir_name}: POSITION_SIZING_MODE={mode} FULL_MARGIN_BUFFER_RATIO={buf}"
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="四所 .env 计仓模式项同步")
|
||||
ap.add_argument("--dry-run", action="store_true", help="仅打印将做的变更")
|
||||
ap.add_argument(
|
||||
"--set-mode",
|
||||
choices=sorted(VALID_MODES),
|
||||
metavar="MODE",
|
||||
help="强制四所 POSITION_SIZING_MODE（须无仓后重启）",
|
||||
)
|
||||
ap.add_argument(
|
||||
"--set-buffer",
|
||||
metavar="RATIO",
|
||||
help=f"强制四所 FULL_MARGIN_BUFFER_RATIO（缺省追加为 {DEFAULT_BUFFER}）",
|
||||
)
|
||||
args = ap.parse_args()
|
||||
if args.set_mode:
|
||||
print(
|
||||
f"注意：将 POSITION_SIZING_MODE 设为 {args.set_mode}，"
|
||||
"请确认交易所无持仓后再 restart。"
|
||||
)
|
||||
for name in INSTANCES:
|
||||
print(
|
||||
sync_one(
|
||||
name,
|
||||
args.dry_run,
|
||||
set_mode=args.set_mode,
|
||||
set_buffer=args.set_buffer,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
将计仓模式相关项写入三所实例 .env（已存在则保留原值，缺失则追加默认值）。
|
||||
|
||||
用法（仓库根目录）：
|
||||
python scripts/sync_four_exchange_position_sizing_env.py
|
||||
python scripts/sync_four_exchange_position_sizing_env.py --dry-run
|
||||
python scripts/sync_four_exchange_position_sizing_env.py --set-mode risk
|
||||
python scripts/sync_four_exchange_position_sizing_env.py --set-mode full_margin
|
||||
|
||||
切换 POSITION_SIZING_MODE 须在交易所无持仓后执行，并 pm2 restart 对应实例。
|
||||
不修改 API 密钥与其它自定义项；若 .env 不存在则跳过（请先从 .env.example 复制）。
|
||||
|
||||
完整说明见 docs/env-sync-scripts.md
|
||||
"""
|
||||
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",
|
||||
)
|
||||
|
||||
COMMENT_POSITION_SIZING = (
|
||||
"# 计仓：risk=以损定仓（默认）；full_margin=合约可用×FULL_MARGIN_BUFFER_RATIO 全仓杠杆（须无仓后重启）"
|
||||
)
|
||||
COMMENT_BUFFER = "# 使用可用资金时的缓冲比例（如0.98代表用98%）"
|
||||
|
||||
DEFAULT_MODE = "risk"
|
||||
DEFAULT_BUFFER = "0.98"
|
||||
VALID_MODES = frozenset({"risk", "full_margin"})
|
||||
|
||||
|
||||
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 _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 _ensure_position_sizing(lines: list[str], *, force_mode: str | None) -> list[str]:
|
||||
if force_mode is not None:
|
||||
if COMMENT_POSITION_SIZING not in lines and not _env_get(lines, "POSITION_SIZING_MODE"):
|
||||
lines = _insert_before(lines, "DAILY_START_CAPITAL", [COMMENT_POSITION_SIZING])
|
||||
return _upsert(lines, "POSITION_SIZING_MODE", force_mode)
|
||||
|
||||
cur = _env_get(lines, "POSITION_SIZING_MODE")
|
||||
if cur is not None:
|
||||
norm = cur.strip().lower()
|
||||
if norm in VALID_MODES and norm != cur:
|
||||
return _upsert(lines, "POSITION_SIZING_MODE", norm)
|
||||
if norm not in VALID_MODES:
|
||||
return _upsert(lines, "POSITION_SIZING_MODE", DEFAULT_MODE)
|
||||
return lines
|
||||
|
||||
block = [COMMENT_POSITION_SIZING, f"POSITION_SIZING_MODE={DEFAULT_MODE}"]
|
||||
return _insert_before(lines, "DAILY_START_CAPITAL", block)
|
||||
|
||||
|
||||
def _ensure_buffer_ratio(lines: list[str], *, force_buffer: str | None) -> list[str]:
|
||||
if force_buffer is not None:
|
||||
if COMMENT_BUFFER not in lines and _env_get(lines, "FULL_MARGIN_BUFFER_RATIO") is None:
|
||||
lines = _insert_before(lines, "BALANCE_REFRESH_SECONDS", [COMMENT_BUFFER])
|
||||
return _upsert(lines, "FULL_MARGIN_BUFFER_RATIO", force_buffer)
|
||||
|
||||
if _env_get(lines, "FULL_MARGIN_BUFFER_RATIO") is not None:
|
||||
return lines
|
||||
|
||||
block = [COMMENT_BUFFER, f"FULL_MARGIN_BUFFER_RATIO={DEFAULT_BUFFER}"]
|
||||
return _insert_before(lines, "BALANCE_REFRESH_SECONDS", block)
|
||||
|
||||
|
||||
def sync_one(
|
||||
dir_name: str,
|
||||
dry_run: bool,
|
||||
*,
|
||||
set_mode: str | None,
|
||||
set_buffer: str | 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）"
|
||||
old_lines = _parse_env(env_path)
|
||||
new_lines = _ensure_buffer_ratio(
|
||||
_ensure_position_sizing(list(old_lines), force_mode=set_mode),
|
||||
force_buffer=set_buffer,
|
||||
)
|
||||
mode = _env_get(new_lines, "POSITION_SIZING_MODE") or DEFAULT_MODE
|
||||
buf = _env_get(new_lines, "FULL_MARGIN_BUFFER_RATIO") or DEFAULT_BUFFER
|
||||
if new_lines == old_lines:
|
||||
return f"OK {dir_name}: POSITION_SIZING_MODE={mode} FULL_MARGIN_BUFFER_RATIO={buf}"
|
||||
if dry_run:
|
||||
return (
|
||||
f"DRY {dir_name}: 将写入 POSITION_SIZING_MODE={mode} "
|
||||
f"FULL_MARGIN_BUFFER_RATIO={buf}"
|
||||
)
|
||||
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")
|
||||
return f"DONE {dir_name}: POSITION_SIZING_MODE={mode} FULL_MARGIN_BUFFER_RATIO={buf}"
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="三所 .env 计仓模式项同步")
|
||||
ap.add_argument("--dry-run", action="store_true", help="仅打印将做的变更")
|
||||
ap.add_argument(
|
||||
"--set-mode",
|
||||
choices=sorted(VALID_MODES),
|
||||
metavar="MODE",
|
||||
help="强制三所 POSITION_SIZING_MODE（须无仓后重启）",
|
||||
)
|
||||
ap.add_argument(
|
||||
"--set-buffer",
|
||||
metavar="RATIO",
|
||||
help=f"强制三所 FULL_MARGIN_BUFFER_RATIO（缺省追加为 {DEFAULT_BUFFER}）",
|
||||
)
|
||||
args = ap.parse_args()
|
||||
if args.set_mode:
|
||||
print(
|
||||
f"注意：将 POSITION_SIZING_MODE 设为 {args.set_mode}，"
|
||||
"请确认交易所无持仓后再 restart。"
|
||||
)
|
||||
for name in INSTANCES:
|
||||
print(
|
||||
sync_one(
|
||||
name,
|
||||
args.dry_run,
|
||||
set_mode=args.set_mode,
|
||||
set_buffer=args.set_buffer,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -1,213 +1,212 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
将每日自动划转相关项写入四所实例 .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 复制）。
|
||||
|
||||
完整说明见 docs/env-sync-scripts.md
|
||||
"""
|
||||
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",
|
||||
)
|
||||
|
||||
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",
|
||||
"TRANSFER_CCY": "USDT",
|
||||
"AUTO_TRANSFER_BJ_HOUR": "8",
|
||||
}
|
||||
|
||||
DEFAULT_AMOUNT = "50"
|
||||
|
||||
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 _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 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 _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）"
|
||||
old_lines = _parse_env(env_path)
|
||||
extra = dict(BINANCE_ONLY) if dir_name == "crypto_monitor_binance" else {}
|
||||
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}: 将更新 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")
|
||||
return f"DONE {dir_name}: ENABLED={enabled} AMOUNT={amt}"
|
||||
|
||||
|
||||
def main():
|
||||
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,
|
||||
set_amount=args.set_amount,
|
||||
enable_auto=True if args.enable_auto_transfer else None,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
将每日自动划转相关项写入三所实例 .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 复制）。
|
||||
|
||||
完整说明见 docs/env-sync-scripts.md
|
||||
"""
|
||||
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",
|
||||
)
|
||||
|
||||
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",
|
||||
"TRANSFER_CCY": "USDT",
|
||||
"AUTO_TRANSFER_BJ_HOUR": "8",
|
||||
}
|
||||
|
||||
DEFAULT_AMOUNT = "50"
|
||||
|
||||
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 _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 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 _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）"
|
||||
old_lines = _parse_env(env_path)
|
||||
extra = dict(BINANCE_ONLY) if dir_name == "crypto_monitor_binance" else {}
|
||||
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}: 将更新 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")
|
||||
return f"DONE {dir_name}: ENABLED={enabled} AMOUNT={amt}"
|
||||
|
||||
|
||||
def main():
|
||||
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,
|
||||
set_amount=args.set_amount,
|
||||
enable_auto=True if args.enable_auto_transfer else None,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user