feat(strategy_manager): 添加策略自动启动的白名单管理

实现全面的白名单管理系统,支持策略自动启动:

- 添加 WhitelistManager 类用于持久化白名单配置存储
- 将白名单功能集成到 StrategyManager 核心模块
- 添加用于白名单 CRUD 操作的 REST API 端点
- 通过 start.py 创建用于白名单管理的 CLI 命令
- 更新前端 UI,添加白名单状态指示器和批量操作功能
- 实现每日 08:58 的自动启动调度
- 为白名单操作添加配置验证和日志记录

同时添加项目文档:
- 代码格式规则(缩进、行长度、导入、命名)
- 文件、变量、常量、函数、类的命名规范
- 受限制文件和目录的保护规则
This commit is contained in:
2026-01-26 01:21:46 +08:00
parent 1b5021d640
commit c0d996f39b
11 changed files with 1952 additions and 99 deletions

View File

@@ -12,6 +12,7 @@ import json # 确保导入json模块
# ==================== 动态路径配置 ====================
from core.path_utils import add_project_root_to_path
from core.whitelist_manager import WhitelistManager
# 添加项目根路径到sys.path
PROJECT_ROOT = add_project_root_to_path()
@@ -34,6 +35,10 @@ class StrategyManager:
# 配置管理器日志
self._setup_logger()
# 初始化白名单管理器
self.whitelist_manager = WhitelistManager()
self.logger.info("📋 白名单管理器已初始化")
self.strategies: Dict[str, Dict[str, Any]] = {}
self.logger.info("🔄 正在加载策略配置...")
self.load_strategies()
@@ -109,15 +114,31 @@ class StrategyManager:
self.logger.error("❌ 加载配置失败 %s: %s", config_file, e, exc_info=True)
def get_status(self) -> Dict[str, Any]:
"""获取完整状态"""
"""获取完整状态(包含白名单信息)"""
self._refresh_status()
return {
# 构建状态数据
status = {
"timestamp": datetime.now().isoformat(),
"total": len(self.strategies),
"running": sum(1 for s in self.strategies.values() if s["status"] == "running"),
"strategies": self.strategies
}
# 添加白名单信息到每个策略
for name, info in status["strategies"].items():
info["in_whitelist"] = self.whitelist_manager.is_in_whitelist(name)
info["whitelist_enabled"] = self.whitelist_manager.is_enabled_in_whitelist(name)
# 添加自动启动状态
auto_start_status = self.whitelist_manager.get_auto_start_status()
status["whitelist_auto_start_today"] = auto_start_status["should_auto_start"]
status["whitelist_last_date"] = auto_start_status["last_auto_start_date"]
status["whitelist_total"] = auto_start_status["whitelist_count"]
status["whitelist_enabled"] = auto_start_status["enabled_count"]
return status
def _refresh_status(self):
"""刷新进程状态 - 双重验证"""
for name, info in self.strategies.items():
@@ -347,6 +368,121 @@ class StrategyManager:
except Exception as e:
self.logger.error("❌ 保存状态失败: %s", e, exc_info=True)
# ==================== 白名单管理方法 ====================
def add_to_whitelist(self, name: str) -> bool:
"""
添加策略到白名单
Args:
name: 策略标识符
Returns:
是否添加成功
"""
if name not in self.strategies:
self.logger.error("❌ 策略不存在: %s", name)
return False
if self.whitelist_manager.add(name, enabled=True):
self.logger.info("✅ 添加到白名单: %s", name)
self._save_status()
return True
return False
def remove_from_whitelist(self, name: str) -> bool:
"""
从白名单移除策略
Args:
name: 策略标识符
Returns:
是否移除成功
"""
if self.whitelist_manager.remove(name):
self.logger.info("✅ 从白名单移除: %s", name)
self._save_status()
return True
return False
def set_whitelist_enabled(self, name: str, enabled: bool) -> bool:
"""
设置策略在白名单中的启用状态
Args:
name: 策略标识符
enabled: 是否启用
Returns:
是否设置成功
"""
if self.whitelist_manager.set_enabled(name, enabled):
self.logger.info("✅ 设置白名单状态: %s -> %s", name, enabled)
self._save_status()
return True
return False
def auto_start_whitelist_strategies(self) -> Dict[str, bool]:
"""
自动启动白名单中所有未运行的策略
一天只执行一次
Returns:
Dict[str, bool]: 每个策略的启动结果
"""
if not self.whitelist_manager.should_auto_start_today():
self.logger.info("⏰ 今天已经执行过自动启动,跳过")
return {}
self.logger.info("🚀 开始执行白名单自动启动...")
results = {}
whitelist = self.whitelist_manager.get_all()
for name, config in whitelist.items():
if not config.get("enabled", True):
self.logger.info("⏭️ 跳过禁用策略: %s", name)
continue
if name not in self.strategies:
self.logger.warning("⚠️ 策略不在系统中: %s", name)
continue
# 检查是否已在运行
if self._is_running(name):
self.logger.info("✅ 策略已在运行: %s", name)
results[name] = True
continue
# 尝试启动
self.logger.info("🚀 启动白名单策略: %s", name)
success = self.start_strategy(name)
# 记录启动结果
results[name] = success
if success:
self.logger.info("✅ 白名单策略启动成功: %s", name)
else:
self.logger.error("❌ 白名单策略启动失败: %s", name)
# 更新日期
self.whitelist_manager.update_last_auto_start_date(
datetime.now().date().isoformat()
)
# 统计结果
success_count = sum(1 for v in results.values() if v)
fail_count = len(results) - success_count
self.logger.info("📊 白名单自动启动完成: 成功 %d, 失败 %d", success_count, fail_count)
return results
def print_status_table(status: Dict[str, Any]):
"""格式化打印状态表格"""

View File

@@ -0,0 +1,345 @@
"""
白名单管理器 - 管理期望自动启动的策略列表
功能:
1. 持久化白名单配置到 JSON 文件
2. 支持添加/移除策略
3. 支持启用/禁用白名单中的策略
4. 跟踪自动启动日期,实现一天只尝试一次
"""
import json
import sys
from pathlib import Path
from typing import Dict, Any, Optional
from datetime import datetime
class WhitelistManager:
"""白名单管理器"""
def __init__(self, config_path: str = "config/whitelist.json"):
"""
初始化白名单管理器
Args:
config_path: 白名单配置文件路径
"""
self.config_path = Path(config_path)
self.data = self._load()
self.logger = self._setup_logger()
def _setup_logger(self):
"""配置日志"""
import logging
logger = logging.getLogger("WhitelistManager")
return logger
def _load(self) -> Dict[str, Any]:
"""
加载白名单配置
Returns:
白名单配置数据
"""
if not self.config_path.exists():
# 返回默认配置
return {
"version": "1.0",
"last_auto_start_date": None,
"strategies": {}
}
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
data = json.load(f)
# 确保数据结构完整
if "version" not in data:
data["version"] = "1.0"
if "strategies" not in data:
data["strategies"] = {}
return data
except Exception as e:
print(f"[WARNING] 加载白名单配置失败: {e}")
return {
"version": "1.0",
"last_auto_start_date": None,
"strategies": {}
}
def _save(self) -> bool:
"""
保存白名单配置
Returns:
是否保存成功
"""
try:
# 确保目录存在
self.config_path.parent.mkdir(exist_ok=True, parents=True)
with open(self.config_path, 'w', encoding='utf-8') as f:
json.dump(self.data, f, indent=2, ensure_ascii=False)
self.logger.debug("白名单配置已保存: %s", self.config_path)
return True
except Exception as e:
self.logger.error("保存白名单配置失败: %s", e)
return False
def add(self, strategy_key: str, enabled: bool = True) -> bool:
"""
添加策略到白名单
Args:
strategy_key: 策略标识符 (如 "DualModeTrendlineHawkesStrategy2_FG")
enabled: 是否启用自动启动
Returns:
是否添加成功False 表示已存在)
"""
if strategy_key in self.data["strategies"]:
self.logger.warning("策略已在白名单中: %s", strategy_key)
return False
self.data["strategies"][strategy_key] = {
"enabled": enabled,
"added_at": datetime.now().isoformat(),
"added_by": "web"
}
if self._save():
self.logger.info("添加策略到白名单: %s (enabled=%s)", strategy_key, enabled)
return True
return False
def remove(self, strategy_key: str) -> bool:
"""
从白名单移除策略
Args:
strategy_key: 策略标识符
Returns:
是否移除成功False 表示不存在)
"""
if strategy_key not in self.data["strategies"]:
self.logger.warning("策略不在白名单中: %s", strategy_key)
return False
del self.data["strategies"][strategy_key]
if self._save():
self.logger.info("从白名单移除策略: %s", strategy_key)
return True
return False
def set_enabled(self, strategy_key: str, enabled: bool) -> bool:
"""
设置策略在白名单中的启用状态
Args:
strategy_key: 策略标识符
enabled: 是否启用
Returns:
是否设置成功
"""
if strategy_key not in self.data["strategies"]:
self.logger.warning("策略不在白名单中: %s", strategy_key)
return False
self.data["strategies"][strategy_key]["enabled"] = enabled
self.data["strategies"][strategy_key]["updated_at"] = datetime.now().isoformat()
if self._save():
self.logger.info("设置白名单策略状态: %s -> %s", strategy_key, enabled)
return True
return False
def get_all(self) -> Dict[str, Dict[str, Any]]:
"""
获取所有白名单策略
Returns:
白名单策略字典 {strategy_key: {enabled, added_at, ...}}
"""
return self.data.get("strategies", {})
def get(self, strategy_key: str) -> Optional[Dict[str, Any]]:
"""
获取单个策略的白名单配置
Args:
strategy_key: 策略标识符
Returns:
策略配置或 None
"""
return self.data["strategies"].get(strategy_key)
def is_in_whitelist(self, strategy_key: str) -> bool:
"""
检查策略是否在白名单中
Args:
strategy_key: 策略标识符
Returns:
是否在白名单中
"""
return strategy_key in self.data.get("strategies", {})
def is_enabled_in_whitelist(self, strategy_key: str) -> bool:
"""
检查策略是否在白名单中且已启用
Args:
strategy_key: 策略标识符
Returns:
是否在白名单中且启用
"""
if strategy_key not in self.data.get("strategies", {}):
return False
return self.data["strategies"][strategy_key].get("enabled", True)
def update_last_auto_start_date(self, date_str: str) -> bool:
"""
更新最后自动启动日期
Args:
date_str: 日期字符串 (YYYY-MM-DD)
Returns:
是否更新成功
"""
self.data["last_auto_start_date"] = date_str
return self._save()
def get_last_auto_start_date(self) -> Optional[str]:
"""
获取最后自动启动日期
Returns:
日期字符串或 None
"""
return self.data.get("last_auto_start_date")
def should_auto_start_today(self) -> bool:
"""
检查今天是否应该自动启动
Returns:
今天是否应该自动启动
"""
today = datetime.now().date().isoformat()
last_date = self.data.get("last_auto_start_date")
if last_date is None:
return True
return last_date != today
def get_auto_start_status(self) -> Dict[str, Any]:
"""
获取自动启动状态
Returns:
自动启动状态信息
"""
today = datetime.now().date().isoformat()
last_date = self.data.get("last_auto_start_date")
return {
"should_auto_start": self.should_auto_start_today(),
"last_auto_start_date": last_date,
"is_first_run_today": last_date != today,
"whitelist_count": len(self.data.get("strategies", {})),
"enabled_count": sum(
1 for s in self.data.get("strategies", {}).values()
if s.get("enabled", True)
)
}
def clear(self) -> bool:
"""
清空白名单(谨慎使用)
Returns:
是否清空成功
"""
self.data["strategies"] = {}
self.data["last_auto_start_date"] = None
return self._save()
# ==================== 单元测试 ====================
if __name__ == "__main__":
import os
# 切换到 strategy_manager 目录
os.chdir(Path(__file__).parent.parent)
# 测试白名单管理器
print("=" * 60)
print("WhitelistManager 单元测试")
print("=" * 60)
# 创建测试用的白名单管理器
test_config_path = "config/whitelist_test.json"
wm = WhitelistManager(test_config_path)
# 测试添加
print("\n[测试] 添加策略到白名单...")
assert wm.add("TestStrategy1_FG") == True
assert wm.add("TestStrategy2_FG", enabled=False) == True
assert wm.add("TestStrategy1_FG") == False # 已存在
# 测试获取
print("\n[测试] 获取白名单...")
all_strategies = wm.get_all()
assert "TestStrategy1_FG" in all_strategies
assert "TestStrategy2_FG" in all_strategies
print(f"白名单策略: {list(all_strategies.keys())}")
# 测试状态检查
print("\n[测试] 状态检查...")
assert wm.is_in_whitelist("TestStrategy1_FG") == True
assert wm.is_in_whitelist("TestStrategy1_FG") == True
assert wm.is_in_whitelist("TestStrategy3_FG") == False
assert wm.is_enabled_in_whitelist("TestStrategy1_FG") == True
assert wm.is_enabled_in_whitelist("TestStrategy2_FG") == False
# 测试设置启用状态
print("\n[测试] 设置启用状态...")
assert wm.set_enabled("TestStrategy2_FG", True) == True
assert wm.is_enabled_in_whitelist("TestStrategy2_FG") == True
assert wm.set_enabled("TestStrategy3_FG", True) == False # 不存在
# 测试移除
print("\n[测试] 移除策略...")
assert wm.remove("TestStrategy2_FG") == True
assert wm.remove("TestStrategy2_FG") == False # 已移除
assert wm.is_in_whitelist("TestStrategy2_FG") == False
# 测试自动启动日期
print("\n[测试] 自动启动日期...")
today = datetime.now().date().isoformat()
assert wm.should_auto_start_today() == True
wm.update_last_auto_start_date(today)
assert wm.should_auto_start_today() == False
wm.update_last_auto_start_date("2024-01-01")
assert wm.should_auto_start_today() == True
# 清理测试文件
print("\n[测试] 清理...")
Path(test_config_path).unlink(missing_ok=True)
print("\n" + "=" * 60)
print("✅ 所有测试通过!")
print("=" * 60)