liaozhaorun/NewQuant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增web界面管理策略

Browse Source
This commit is contained in:
liaozhaorun
2025-11-21 16:08:03 +08:00
parent 4b7ec4e564
commit 218ca5f533
11 changed files with 747 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

218
strategy_manager/core/manager.py
View File

@@ -1,12 +1,14 @@
import os
import sys
import json
import time
import psutil
import subprocess
import logging
from pathlib import Path
from typing import Dict, Any, List
from datetime import datetime
import importlib.util
import json # 确保导入json模块
# ==================== 动态路径配置 ====================
from core.path_utils import add_project_root_to_path
@@ -29,12 +31,32 @@ class StrategyManager:
self.logs_dir.mkdir(exist_ok=True)
self.pid_dir.mkdir(exist_ok=True)
# 配置管理器日志
self._setup_logger()
self.strategies: Dict[str, Dict[str, Any]] = {}
self.logger.info("🔄 正在加载策略配置...")
self.load_strategies()
self.logger.info("✅ 策略加载完成，共发现 %d 个策略", len(self.strategies))
def _setup_logger(self):
"""配置管理器日志"""
log_file = self.logs_dir / "manager.log"
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler(log_file, encoding='utf-8'),
logging.StreamHandler(sys.stdout)
],
force=True
)
self.logger = logging.getLogger("StrategyManager")
def _load_main_config(self, config_path: str) -> Dict[str, Any]:
path = Path(config_path)
if not path.exists():
self.logger.warning("⚠️ 主配置文件不存在，使用默认配置")
return {
"logs_dir": "logs",
"status_file": "status.json",
@@ -44,16 +66,23 @@ class StrategyManager:
return json.load(f)
def load_strategies(self):
"""递归扫描 strategies/ 目录，查找 .config 文件"""
"""递归扫描 strategies/ 目录，查找 .py 配置文件"""
self.strategies = {}
if not self.strategies_dir.exists():
print("[ERROR] 策略配置目录不存在: {}".format(self.strategies_dir))
self.logger.error("❌ 策略配置目录不存在: %s", self.strategies_dir)
return
for config_file in self.strategies_dir.rglob("*.config"):
for config_file in self.strategies_dir.rglob("*.py"):
try:
with open(config_file, 'r', encoding='utf-8') as f:
config = json.load(f)
spec = importlib.util.spec_from_file_location(
f"config_{config_file.stem}", config_file
)
config_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(config_module)
if not hasattr(config_module, 'CONFIG'):
raise ValueError("配置文件缺少 CONFIG 变量")
config = config_module.CONFIG
required = ['name', 'strategy_class', 'enabled', 'engine_params', 'strategy_params']
for field in required:
@@ -75,8 +104,9 @@ class StrategyManager:
"started_at": None,
"uptime": None
}
self.logger.info("📄 加载配置: %s", config_file)
except Exception as e:
print("[ERROR] 加载配置失败 {}: {}".format(config_file, e))
self.logger.error("❌ 加载配置失败 %s: %s", config_file, e, exc_info=True)
def get_status(self) -> Dict[str, Any]:
"""获取完整状态"""
@@ -89,12 +119,7 @@ class StrategyManager:
}
def _refresh_status(self):
"""
刷新进程状态 - 双重验证
1. 检查PID文件是否存在
2. 检查进程是否存在
3. 验证进程名是否为python（防止PID复用）
"""
"""刷新进程状态 - 双重验证"""
for name, info in self.strategies.items():
pid_file = self.pid_dir / "{}.pid".format(name)
@@ -103,28 +128,23 @@ class StrategyManager:
with open(pid_file, 'r') as f:
pid = int(f.read().strip())
# 双重验证
if psutil.pid_exists(pid):
try:
proc = psutil.Process(pid)
# 验证进程名是否包含python
if "python" in proc.name().lower():
# 验证成功，更新为运行中
info["status"] = "running"
info["pid"] = pid
if info["started_at"]:
started = datetime.fromisoformat(info["started_at"])
uptime = datetime.now() - started
info["uptime"] = str(uptime).split('.')[0]
continue # 跳过清理逻辑
continue
except (psutil.NoSuchProcess, psutil.AccessDenied):
# 进程已死或无权访问，继续清理
pass
# PID不存在或验证失败，清理
self._cleanup_stopped_strategy(name, pid_file)
except Exception as e:
print("[WARNING] 刷新状态失败 {}: {}".format(name, e))
self.logger.warning("⚠️ 刷新状态失败 %s: %s", name, e)
self._cleanup_stopped_strategy(name, pid_file)
else:
info["status"] = "stopped"
@@ -133,13 +153,8 @@ class StrategyManager:
info["uptime"] = None
def _is_running(self, name: str) -> bool:
"""
检查策略是否运行中 - 实时刷新状态
确保与status命令结果一致
"""
# 先刷新状态确保最新
"""检查策略是否运行中"""
self._refresh_status()
info = self.strategies[name]
if not info["pid"]:
return False
@@ -152,7 +167,6 @@ class StrategyManager:
with open(pid_file, 'r') as f:
pid = int(f.read().strip())
# 双重验证
if psutil.pid_exists(pid):
try:
proc = psutil.Process(pid)
@@ -163,15 +177,74 @@ class StrategyManager:
except:
return False
def start_strategy(self, name: str) -> bool:
"""启动单个策略（守护进程模式）"""
if name not in self.strategies:
self.logger.error("❌ 策略不存在: %s", name)
return False
if self._is_running(name):
self.logger.warning("⚠️ 策略已在运行: %s", name)
return False
info = self.strategies[name]
config_file = Path(info["config_file"])
# 创建策略专属日志目录
strategy_log_dir = self.logs_dir / info["strategy_name"]
strategy_log_dir.mkdir(exist_ok=True, parents=True)
log_file = strategy_log_dir / "{}.log".format(info["symbol"])
self.logger.info("\n" + "=" * 80)
self.logger.info("🚀 启动策略: %s", name)
self.logger.info("📄 配置文件: %s", config_file)
self.logger.info("📝 日志文件: %s", log_file)
try:
# 启动子进程 - 关键修改：脱离终端会话
with open(log_file, 'a') as f:
# 使用 start_new_session=True 创建新会话，防止终端关闭影响
# stdin 重定向到 /dev/null，完全脱离终端
with open(os.devnull, 'r') as devnull:
process = subprocess.Popen(
[sys.executable, "launcher.py", "--config", str(config_file)],
stdout=f,
stderr=subprocess.STDOUT,
stdin=devnull, # 关键：从/dev/null读取输入
# start_new_session=True, # 关键：创建新会话，脱离终端
cwd=Path.cwd()
)
# 更新状态
info["pid"] = process.pid
info["status"] = "running"
info["started_at"] = datetime.now().isoformat()
info["uptime"] = "00:00:00"
# 保存PID文件
pid_file = self.pid_dir / "{}.pid".format(name)
with open(pid_file, 'w') as f:
f.write(str(process.pid))
self._save_status()
self.logger.info("✅ 启动成功! PID: %d", process.pid)
self.logger.info("ℹ️ 该进程已脱离终端会话，关闭窗口不会停止策略")
self.logger.info("=" * 80)
return True
except Exception as e:
self.logger.error("❌ 启动失败: %s", e, exc_info=True)
self._cleanup_stopped_strategy(name, self.pid_dir / "{}.pid".format(name))
return False
def stop_strategy(self, name: str, timeout: int = 30) -> bool:
"""停止单个策略"""
if name not in self.strategies:
print("[ERROR] 策略不存在: {}".format(name))
self.logger.error("❌ 策略不存在: %s", name)
return False
# 再次检查状态（确保最新）
if not self._is_running(name):
print("[WARNING] 策略未运行: {}".format(name))
self.logger.warning("⚠️ 策略未运行: %s", name)
return False
info = self.strategies[name]
@@ -180,40 +253,43 @@ class StrategyManager:
pid = info["pid"]
process = psutil.Process(pid)
print("\n[INFO] 正在停止: {} (PID: {})...".format(name, pid))
self.logger.info("\n" + "=" * 80)
self.logger.info("⏹️ 正在停止策略: %s (PID: %d)", name, pid)
# 优雅终止
process.terminate()
try:
process.wait(timeout=timeout)
print("[SUCCESS] 已停止: {}".format(name))
self.logger.info("✅ 已优雅停止: %s", name)
except psutil.TimeoutExpired:
print("[WARNING] 超时，强制终止: {}".format(name))
self.logger.warning("⏱️ 超时，强制终止: %s", name)
process.kill()
process.wait()
# 清理状态
self._cleanup_stopped_strategy(name, self.pid_dir / "{}.pid".format(name))
self._save_status()
self.logger.info("=" * 80)
return True
except Exception as e:
print("[ERROR] 停止失败 {}: {}".format(name, e))
self.logger.error("❌ 停止失败 %s: %s", name, e, exc_info=True)
return False
def restart_strategy(self, name: str) -> bool:
"""重启策略"""
print("\n[INFO] 正在重启: {}".format(name))
self.logger.info("\n" + "=" * 80)
self.logger.info("🔄 正在重启: %s", name)
self.stop_strategy(name)
time.sleep(2)
return self.start_strategy(name)
def start_all(self):
"""启动所有启用的策略"""
print("\n" + "=" * 100)
print("正在启动所有启用的策略...")
print("=" * 100)
self.logger.info("\n" + "=" * 100)
self.logger.info("🚀 正在启动所有启用的策略...")
self.logger.info("=" * 100)
started = []
for name, info in self.strategies.items():
@@ -221,15 +297,15 @@ class StrategyManager:
if self.start_strategy(name):
started.append(name)
print("\n[SUCCESS] 成功启动 {} 个策略".format(len(started)))
self.logger.info("\n✅ 成功启动 %d 个策略", len(started))
if started:
print("策略: {}".format(", ".join(started)))
self.logger.info("📋 策略: %s", ", ".join(started))
def stop_all(self):
"""停止所有运行的策略"""
print("\n" + "=" * 100)
print("正在停止所有运行的策略...")
print("=" * 100)
self.logger.info("\n" + "=" * 100)
self.logger.info("⏹️ 正在停止所有运行的策略...")
self.logger.info("=" * 100)
stopped = []
for name in self.strategies.keys():
@@ -237,9 +313,9 @@ class StrategyManager:
if self.stop_strategy(name):
stopped.append(name)
print("\n[SUCCESS] 成功停止 {} 个策略".format(len(stopped)))
self.logger.info("\n✅ 成功停止 %d 个策略", len(stopped))
if stopped:
print("策略: {}".format(", ".join(stopped)))
self.logger.info("📋 策略: %s", ", ".join(stopped))
def _cleanup_stopped_strategy(self, name: str, pid_file: Path):
"""清理已停止的策略状态"""
@@ -251,10 +327,25 @@ class StrategyManager:
info["uptime"] = None
def _save_status(self):
"""状态持久化"""
status = self.get_status()
with open(self.status_file, 'w') as f:
json.dump(status, f, indent=2, ensure_ascii=False)
"""状态持久化（修复：排除不可序列化的config字段）"""
try:
status = self.get_status()
# 创建JSON安全的版本（排除config字段，因为它可能包含timedelta等不可序列化对象）
status_for_json = status.copy()
status_for_json["strategies"] = {}
for name, info in status["strategies"].items():
# 复制所有字段除了config
strategy_info = {k: v for k, v in info.items() if k != "config"}
status_for_json["strategies"][name] = strategy_info
with open(self.status_file, 'w') as f:
json.dump(status_for_json, f, indent=2, ensure_ascii=False)
self.logger.debug("💾 状态已保存到 %s", self.status_file)
except Exception as e:
self.logger.error("❌ 保存状态失败: %s", e, exc_info=True)
def print_status_table(status: Dict[str, Any]):
@@ -284,4 +375,31 @@ def print_status_table(status: Dict[str, Any]):
name, info['config']['name'], status_text, pid_text, uptime_text, started_text
))
print("=" * 130)
print("=" * 130)
if __name__ == "__main__":
manager = StrategyManager()
if len(sys.argv) > 1:
command = sys.argv[1]
if command == "status":
print_status_table(manager.get_status())
elif command == "start-all":
manager.start_all()
elif command == "stop-all":
manager.stop_all()
elif command.startswith("start:"):
name = command.split(":", 1)[1]
manager.start_strategy(name)
elif command.startswith("stop:"):
name = command.split(":", 1)[1]
manager.stop_strategy(name)
elif command.startswith("restart:"):
name = command.split(":", 1)[1]
manager.restart_strategy(name)
else:
print("未知命令:", command)
print("用法: python manager.py [status|start-all|stop-all|start:NAME|stop:NAME|restart:NAME]")
else:
print("用法: python manager.py [status|start-all|stop-all|start:NAME|stop:NAME|restart:NAME]")