diff --git a/scripts/apply_recommended_config.py b/scripts/apply_recommended_config.py new file mode 100644 index 0000000..e90337f --- /dev/null +++ b/scripts/apply_recommended_config.py @@ -0,0 +1,211 @@ +#!/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]] = { + "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() +