#!/usr/bin/env python3 """ 一键批量应用全局策略配置（推荐配置 / 自定义配置）。 用途： - 避免在前端逐项手动修改配置； - 支持默认推荐值、外部 JSON 覆盖、命令行临时覆盖； - 自动写入 global_strategy_config，并尽量同步到 Redis 配置缓存。 示例： python scripts/apply_recommended_config.py python scripts/apply_recommended_config.py --dry-run python scripts/apply_recommended_config.py --from config/my_batch.json python scripts/apply_recommended_config.py --set MIN_SIGNAL_STRENGTH=9 --set SHADOW_MODE_AUTO_APPLY=true """ from __future__ import annotations import argparse import json import os import sys from pathlib import Path from typing import Any, Dict, List PROJECT_ROOT = Path(__file__).resolve().parent.parent BACKEND_DIR = PROJECT_ROOT / "backend" if str(BACKEND_DIR) not in sys.path: sys.path.insert(0, str(BACKEND_DIR)) if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) # 默认推荐配置（可按你的风格长期维护） DEFAULT_RECOMMENDED: Dict[str, Dict[str, Any]] = { # 空=不限制；非空则仅名单内合约自动下单（窄宇宙），见 config/strategy_narrow_universe.json "AUTO_TRADE_SYMBOL_WHITELIST": {"value": "", "type": "string", "category": "strategy"}, "AUTO_TRADE_ONLY_TRENDING": {"value": True, "type": "boolean", "category": "strategy"}, "BETA_FILTER_ENABLED": {"value": True, "type": "boolean", "category": "strategy"}, "BETA_FILTER_THRESHOLD": {"value": -0.01, "type": "number", "category": "strategy"}, "MIN_SIGNAL_STRENGTH": {"value": 8, "type": "number", "category": "strategy"}, "RANGING_MARKET_SIGNAL_BOOST": {"value": 2, "type": "number", "category": "strategy"}, "NO_OPEN_HOURS_BJ": {"value": "1,2,3,4,5,6,8,9,17,18,19", "type": "string", "category": "risk"}, "MANUAL_BLOCKED_SYMBOLS": {"value": "", "type": "string", "category": "strategy"}, "MANUAL_REDUCED_SYMBOLS": { "value": "HYPEUSDT,XANUSDT,WIFUSDT,ANKRUSDT,RIVERUSDT,POLYXUSDT,PIPPINUSDT,AINUSDT,ANIMEUSDT,COSUSDT,GUNUSDT,HYPERUSDT", "type": "string", "category": "strategy", }, "MANUAL_REDUCED_SYMBOL_POSITION_FACTOR": {"value": 0.5, "type": "number", "category": "strategy"}, "MANUAL_REDUCED_SYMBOL_SIGNAL_BOOST": {"value": 1, "type": "number", "category": "strategy"}, "ENTRY_PULLBACK_FILTER_ENABLED": {"value": True, "type": "boolean", "category": "strategy"}, "ENTRY_PULLBACK_INTERVAL": {"value": "", "type": "string", "category": "strategy"}, "ENTRY_PULLBACK_LOOKBACK_BARS": {"value": 24, "type": "number", "category": "strategy"}, "ENTRY_PULLBACK_MIN_BARS": {"value": 5, "type": "number", "category": "strategy"}, "ENTRY_PULLBACK_MAX_LONG_IN_RANGE": {"value": 0.58, "type": "number", "category": "strategy"}, "ENTRY_PULLBACK_MIN_SHORT_IN_RANGE": {"value": 0.42, "type": "number", "category": "strategy"}, "SHADOW_MODE_AUTO_APPLY": {"value": True, "type": "boolean", "category": "strategy"}, "SHADOW_MODE_MIN_CONFIDENCE": {"value": 0.7, "type": "number", "category": "strategy"}, "SHADOW_MODE_INCREASE_LEVERAGE_MULT": {"value": 1.3, "type": "number", "category": "strategy"}, "SHADOW_MODE_DECREASE_LEVERAGE_MULT": {"value": 0.5, "type": "number", "category": "strategy"}, "SHADOW_MODE_SUGGESTIONS_PATH": {"value": "config/current_suggestions.json", "type": "string", "category": "strategy"}, "SHADOW_MODE_TRACKING_PATH": {"value": "config/shadow_mode_tracking.json", "type": "string", "category": "strategy"}, } def parse_cli_value(raw: str) -> Any: v = raw.strip() low = v.lower() if low in ("true", "false"): return low == "true" try: if "." in v: return float(v) return int(v) except ValueError: return v def infer_type(v: Any) -> str: if isinstance(v, bool): return "boolean" if isinstance(v, (int, float)): return "number" return "string" def load_external_json(path: Path) -> Dict[str, Dict[str, Any]]: with open(path, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, list): out: Dict[str, Dict[str, Any]] = {} for item in data: if not isinstance(item, dict) or "key" not in item: continue key = str(item["key"]).strip() if not key: continue out[key] = { "value": item.get("value"), "type": item.get("type") or infer_type(item.get("value")), "category": item.get("category") or "strategy", "description": item.get("description"), } return out if isinstance(data, dict): out = {} for key, val in data.items(): if isinstance(val, dict) and "value" in val: out[key] = { "value": val.get("value"), "type": val.get("type") or infer_type(val.get("value")), "category": val.get("category") or "strategy", "description": val.get("description"), } else: out[key] = { "value": val, "type": infer_type(val), "category": "strategy", "description": None, } return out raise ValueError("外部 JSON 必须是对象或数组") def main() -> None: parser = argparse.ArgumentParser(description="批量应用全局策略配置") parser.add_argument("--from", dest="from_path", default="", help="外部 JSON 配置文件路径") parser.add_argument( "--set", action="append", default=[], help="临时覆盖项，格式 KEY=VALUE，可重复多次", ) parser.add_argument("--dry-run", action="store_true", help="只打印，不落库") parser.add_argument("--updated-by", default="apply_recommended_config", help="写入 updated_by") args = parser.parse_args() merged = dict(DEFAULT_RECOMMENDED) if args.from_path: from_path = Path(args.from_path) if not from_path.is_absolute(): from_path = PROJECT_ROOT / from_path if not from_path.is_file(): raise FileNotFoundError(f"未找到配置文件: {from_path}") ext = load_external_json(from_path) merged.update(ext) for item in args.set: if "=" not in item: raise ValueError(f"--set 参数格式错误: {item}（应为 KEY=VALUE）") k, raw = item.split("=", 1) key = k.strip() value = parse_cli_value(raw) if not key: raise ValueError(f"--set key 不能为空: {item}") base = merged.get(key, {}) merged[key] = { "value": value, "type": base.get("type") or infer_type(value), "category": base.get("category") or "strategy", "description": base.get("description"), } print(f"将应用 {len(merged)} 个配置项（dry_run={args.dry_run}）") for key in sorted(merged.keys()): meta = merged[key] print(f" - {key} = {meta.get('value')} [{meta.get('type')}/{meta.get('category')}]") if args.dry_run: return from database.models import GlobalStrategyConfig mgr = None try: import config_manager if hasattr(config_manager, "GlobalStrategyConfigManager"): mgr = config_manager.GlobalStrategyConfigManager() except Exception: mgr = None ok, fail = 0, 0 for key, meta in merged.items(): try: GlobalStrategyConfig.set( key, meta.get("value"), meta.get("type") or infer_type(meta.get("value")), meta.get("category") or "strategy", meta.get("description"), updated_by=args.updated_by, ) if mgr is not None: try: mgr._set_to_redis(key, meta.get("value")) mgr._cache[key] = meta.get("value") except Exception: pass ok += 1 except Exception as e: fail += 1 print(f"[失败] {key}: {e}") print(f"\n完成：成功 {ok}，失败 {fail}") if fail > 0: sys.exit(1) if __name__ == "__main__": main()