liaozhaorun
e88ba5bcf9
- 新增 config_models.py: 使用 Pydantic 提供强类型配置校验 - QMTConfig, QMTTerminalConfig, StrategyConfig 等数据模型 - 支持 slots/percentage 两种下单模式 - 兼容旧版配置格式迁移 - 新增 validate_config.py: 配置检测 CLI 工具 - 重构 TradingUnit 和 MultiEngineManager 使用新配置模型 - 新增百分比模式买卖逻辑 (_execute_percentage_buy/sell) - 完善日志记录和错误处理 - 删除 TODO_FIX.md: 清理已完成的缺陷修复任务清单
122 lines
3.3 KiB
Python
122 lines
3.3 KiB
Python
# coding: utf-8
|
||
"""
|
||
QMT 配置检测工具
|
||
|
||
用于检测配置文件是否正确，包括：
|
||
- 必要字段是否缺失
|
||
- 字段类型是否正确
|
||
- 业务规则是否满足（如 slots 模式必须配置 total_slots）
|
||
- 策略引用的终端是否存在
|
||
- 路径是否存在
|
||
- 是否存在未知配置项
|
||
|
||
使用方法:
|
||
python validate_config.py [config_path]
|
||
|
||
如果不指定 config_path，默认检测 config.json
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
|
||
# 直接导入 config_models 避免依赖 redis
|
||
exec(
|
||
open(
|
||
os.path.join(os.path.dirname(__file__), "config_models.py"), encoding="utf-8"
|
||
).read()
|
||
)
|
||
|
||
|
||
def validate_config(config_path: str = "config.json") -> bool:
|
||
"""
|
||
检测配置文件
|
||
|
||
Args:
|
||
config_path: 配置文件路径
|
||
|
||
Returns:
|
||
bool: 检测是否通过
|
||
"""
|
||
print(f"正在检测配置文件: {config_path}")
|
||
print("-" * 50)
|
||
|
||
try:
|
||
config = load_config(config_path)
|
||
|
||
# 检测通过，显示配置摘要
|
||
print("[OK] 配置检测通过！")
|
||
print()
|
||
print("配置摘要:")
|
||
print(f" - 终端数量: {len(config.qmt_terminals)}")
|
||
print(f" - 策略数量: {len(config.strategies)}")
|
||
print()
|
||
|
||
# 显示终端信息
|
||
if config.qmt_terminals:
|
||
print("终端配置:")
|
||
for terminal in config.qmt_terminals:
|
||
print(f" [{terminal.qmt_id}] {terminal.alias}")
|
||
print(f" 路径: {terminal.path}")
|
||
print(f" 账号: {terminal.account_id}")
|
||
print(f" 类型: {terminal.account_type}")
|
||
# 显示该终端关联的策略
|
||
strategies = config.get_strategies_by_terminal(terminal.qmt_id)
|
||
if strategies:
|
||
print(f" 关联策略: {', '.join(strategies)}")
|
||
print()
|
||
|
||
# 显示策略信息
|
||
if config.strategies:
|
||
print("策略配置:")
|
||
for name, strat in config.strategies.items():
|
||
print(f" [{name}]")
|
||
print(f" 关联终端: {strat.qmt_id}")
|
||
print(f" 下单模式: {strat.order_mode}")
|
||
if strat.order_mode == "slots":
|
||
print(f" 总槽位: {strat.total_slots}")
|
||
print(f" 权重: {strat.weight}")
|
||
print()
|
||
|
||
# 显示自动重连配置
|
||
if config.auto_reconnect:
|
||
print("自动重连配置:")
|
||
print(f" 启用: {config.auto_reconnect.enabled}")
|
||
print(f" 重连时间: {config.auto_reconnect.reconnect_time}")
|
||
print()
|
||
|
||
return True
|
||
|
||
except ConfigError as e:
|
||
print("[FAIL] 配置检测失败！")
|
||
print()
|
||
print("错误详情:")
|
||
print(f" {e}")
|
||
print()
|
||
print("请检查配置文件后重试。")
|
||
return False
|
||
except FileNotFoundError:
|
||
print(f"[FAIL] 配置文件不存在: {config_path}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"[FAIL] 未知错误: {e}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
# 获取配置文件路径
|
||
if len(sys.argv) > 1:
|
||
config_path = sys.argv[1]
|
||
else:
|
||
config_path = "config.json"
|
||
|
||
# 执行检测
|
||
success = validate_config(config_path)
|
||
|
||
# 返回退出码
|
||
sys.exit(0 if success else 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|