crypto_monitor_user/scripts/sync_common_trading_env.py

#!/usr/bin/env python3
"""
将三所共用的交易/关键位/轮询 env 写入币安,OKX 的 .env(缺失则追加,不覆盖已有值).

以 Gate .env.example 为基准;Gate 自身也可运行以补缺失项.

用法(仓库根目录):
  python scripts/sync_common_trading_env.py
  python scripts/sync_common_trading_env.py --dry-run
  python scripts/sync_common_trading_env.py --instances crypto_monitor_okx

修改后须 pm2 restart 对应实例.说明见 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__)))

DEFAULT_INSTANCES = (
    "crypto_monitor_binance",
    "crypto_monitor_okx",
)

# 与 crypto_monitor_gate/.env.example 对齐(不含 GATE_* / 各所 API 密钥)
SHARED_DEFAULTS: dict[str, str] = {
    "TRADING_DAY_RESET_OPEN_GUARD_ENABLED": "true",
    "KEY_CONFIRM_BREAKOUT_BAR": "-2",
    "KEY_CONFIRM_BAR": "-1",
    "KEY_VOLUME_MA_BARS": "20",
    "KEY_VOLUME_RATIO_MIN": "1.3",
    "KEY_BREAKOUT_AMP_MIN_PCT": "0.03",
    "KEY_BREAKOUT_AMP_MAX_PCT": "0.5",
    "KEY_ALERT_MAX_TIMES": "3",
    "KEY_ALERT_INTERVAL_MINUTES": "5",
    "KEY_DAILY_VOLUME_RANK_MAX": "30",
    "KEY_AUTO_MIN_PLANNED_RR": "1.5",
    "KEY_STOP_OUTSIDE_BREAKOUT_PCT": "0.5",
    "KEY_TREND_STOP_OUTSIDE_PCT": "1",
    "MAX_ACTIVE_POSITIONS": "1",
    "MANUAL_MIN_PLANNED_RR": "1.4",
    "KEY_SIZING_USE_ZERO_POSITION_SNAPSHOT": "true",
    "DAILY_OPEN_ALERT_THRESHOLD": "5",
    "DAILY_OPEN_HARD_LIMIT": "0",
    "BALANCE_REFRESH_SECONDS": "60",
    "PRICE_REFRESH_SECONDS": "5",
    "MONITOR_POLL_SECONDS": "3",
    "RECONCILE_STARTUP_GRACE_SEC": "90",
    "RECONCILE_FLAT_CONFIRM_POLLS": "3",
    "FULL_MARGIN_BUFFER_RATIO": "0.98",
    "WECHAT_TIMEOUT_SECONDS": "10",
    "AI_TIMEOUT_SECONDS": "120",
}

# 仅当某实例 .env 缺少 FORCE_CLOSE_* 时补默认:
# Gate 默认开 0 点强制清仓;币安/OKX 默认关.已有手调值绝不覆盖.
FORCE_CLOSE_POLICY: dict[str, dict[str, str]] = {
    "crypto_monitor_gate": {
        "FORCE_CLOSE_ENABLED": "true",
        "FORCE_CLOSE_BJ_HOUR": "0",
    },
    "crypto_monitor_binance": {
        "FORCE_CLOSE_ENABLED": "false",
        "FORCE_CLOSE_BJ_HOUR": "0",
    },
    "crypto_monitor_okx": {
        "FORCE_CLOSE_ENABLED": "false",
        "FORCE_CLOSE_BJ_HOUR": "0",
    },
}


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: list[str] = []
    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 sync_one(dir_name: str, *, dry_run: bool, force: bool) -> bool:
    path = os.path.join(REPO, dir_name, ".env")
    if not os.path.isfile(path):
        print(f"skip (no .env): {dir_name}")
        return False
    lines = _parse_env(path)
    added: list[str] = []
    for key, val in SHARED_DEFAULTS.items():
        cur = _env_get(lines, key)
        if cur is None or (force and cur != val):
            lines = _upsert(lines, key, val)
            added.append(key)
    if not added:
        print(f"ok (unchanged): {dir_name}")
        return False
    print(f"update: {dir_name}")
    for key in added:
        print(f"  + {key}={SHARED_DEFAULTS[key]}")
    if not dry_run:
        text = "\n".join(lines).rstrip() + "\n"
        with open(path, "w", encoding="utf-8", newline="\n") as f:
            f.write(text)
    return True


def apply_force_close_policy(*, dry_run: bool) -> bool:
    """仅在 FORCE_CLOSE_* 缺失时补默认值;已有手调值绝不覆盖."""
    any_changed = False
    for dir_name, values in FORCE_CLOSE_POLICY.items():
        path = os.path.join(REPO, dir_name, ".env")
        if not os.path.isfile(path):
            print(f"skip (no .env): {dir_name}")
            continue
        lines = _parse_env(path)
        added_keys: list[str] = []
        for key, val in values.items():
            cur = _env_get(lines, key)
            if cur is None:
                lines = _upsert(lines, key, val)
                added_keys.append(key)
        if not added_keys:
            print(f"ok (force-close unchanged): {dir_name}")
            continue
        any_changed = True
        print(f"force-close fill-missing: {dir_name}")
        for key in added_keys:
            print(f"  + {key}={values[key]}")
        if not dry_run:
            text = "\n".join(lines).rstrip() + "\n"
            with open(path, "w", encoding="utf-8", newline="\n") as f:
                f.write(text)
    return any_changed


def main() -> None:
    ap = argparse.ArgumentParser(description="同步币安/OKX 共用 trading env(缺失项追加)")
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--force", action="store_true", help="覆盖已有值(慎用)")
    ap.add_argument(
        "--apply-force-close-policy",
        action="store_true",
        help="仅补全缺失的 FORCE_CLOSE_* 默认值(不覆盖手调)",
    )
    ap.add_argument(
        "--instances",
        nargs="+",
        metavar="DIR",
        help="默认 crypto_monitor_binance crypto_monitor_okx",
    )
    args = ap.parse_args()

    instances = tuple(args.instances) if args.instances else DEFAULT_INSTANCES
    any_changed = False
    for inst in instances:
        if sync_one(inst, dry_run=args.dry_run, force=args.force):
            any_changed = True
    if args.apply_force_close_policy:
        if apply_force_close_policy(dry_run=args.dry_run):
            any_changed = True
    if args.dry_run and any_changed:
        print("(dry-run, 未写入)")


if __name__ == "__main__":
    main()