chore: add script to sync position sizing env across four exchanges
sync_four_exchange_position_sizing_env.py appends missing POSITION_SIZING_MODE and FULL_MARGIN_BUFFER_RATIO, optional --set-mode/--set-buffer; docs updated. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
@@ -0,0 +1,178 @@
|
||||
#!/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 复制）。
|
||||
"""
|
||||
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()
|
||||
Reference in New Issue
Block a user