crypto_monitor/scripts/sync_four_exchange_position_sizing_env.py

#!/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()