a08b8d31c6
Introduced a new configuration option for an auto trade symbol whitelist, allowing only specified contracts to execute automatic trades. Updated relevant components to support this feature, enhancing control over automated trading strategies. The whitelist can be configured to restrict or allow trades based on user-defined criteria, improving risk management and trading precision.
214 lines
8.3 KiB
Python
214 lines
8.3 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]] = {
|
||
# 空=不限制；非空则仅名单内合约自动下单（窄宇宙），见 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()
|
||
|