dekun
142 lines
4.5 KiB
Python
142 lines
4.5 KiB
Python
#!/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",
|
|
"FORCE_CLOSE_BJ_HOUR": "0",
|
|
"FORCE_CLOSE_ENABLED": "false",
|
|
"WECHAT_TIMEOUT_SECONDS": "10",
|
|
"AI_TIMEOUT_SECONDS": "120",
|
|
}
|
|
|
|
|
|
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 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(
|
|
"--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.dry_run and any_changed:
|
|
print("(dry-run, 未写入)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|