212 lines
8.0 KiB
Python
212 lines
8.0 KiB
Python
#!/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()
|
||
|