NewQuant/strategy_manager/start.py

import sys
import argparse
from pathlib import Path
from core.manager import StrategyManager, print_status_table


def main():
    parser = argparse.ArgumentParser(
        description="策略管理系统 - 批量管理你的交易策略",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  # 查看所有策略状态
  python start.py status

  # 启动所有策略
  python start.py start --all

  # 启动单个策略（使用标识符）
  python start.py start -n DualModeTrendlineHawkesStrategy2_FG

  # 停止单个策略
  python start.py stop -n DualModeTrendlineHawkesStrategy2_FG

  # 重启策略
  python start.py restart -n DualModeTrendlineHawkesStrategy2_FG

  # 查看日志（最近50行）
  python start.py logs -n DualModeTrendlineHawkesStrategy2_FG -t 50

  # ========== 白名单管理 ==========
  # 查看白名单
  python start.py whitelist

  # 添加策略到白名单
  python start.py whitelist add -n DualModeTrendlineHawkesStrategy2_FG

  # 从白名单移除策略
  python start.py whitelist remove -n DualModeTrendlineHawkesStrategy2_FG

  # 启用白名单中的策略
  python start.py whitelist enable -n DualModeTrendlineHawkesStrategy2_FG

  # 禁用白名单中的策略
  python start.py whitelist disable -n DualModeTrendlineHawkesStrategy2_FG

  # 手动触发白名单自动启动
  python start.py whitelist auto-start
        """
    )

    parser.add_argument("action", choices=["start", "stop", "restart", "status", "logs", "whitelist"])
    parser.add_argument("-n", "--name", help="策略标识符（策略名_品种）")
    parser.add_argument("-a", "--all", action="store_true", help="对所有策略执行操作")
    parser.add_argument("-c", "--config", default="config/main.json", help="主配置文件路径")
    parser.add_argument("-t", "--tail", type=int, default=30, help="查看日志末尾行数")

    # 白名单子命令
    whitelist_group = parser.add_argument_group("白名单操作")
    whitelist_group.add_argument("whitelist_action", choices=["list", "add", "remove", "enable", "disable", "auto-start"],
                                 help="白名单操作动作", nargs="?")

    args = parser.parse_args()
    manager = StrategyManager(args.config)

    # 白名单管理
    if args.action == "whitelist":
        if args.whitelist_action == "list" or args.whitelist_action is None:
            # 列出白名单
            print("\n" + "=" * 80)
            print("📋 白名单列表")
            print("=" * 80)

            whitelist = manager.whitelist_manager.get_all()
            auto_status = manager.whitelist_manager.get_auto_start_status()

            print(f"白名单策略总数: {auto_status['whitelist_count']}")
            print(f"已启用策略数: {auto_status['enabled_count']}")
            print(f"今天已自动启动: {'是' if not auto_status['should_auto_start'] else '否'}")
            print(f"上次自动启动日期: {auto_status['last_auto_start_date'] or '从未'}")
            print("-" * 80)

            if not whitelist:
                print("白名单为空")
            else:
                print(f"{'策略标识':<45} {'状态':<10} {'添加时间'}")
                print("-" * 80)
                for name, config in whitelist.items():
                    status = "启用" if config.get("enabled", True) else "禁用"
                    added_at = config.get("added_at", "")[:19]
                    print(f"{name:<45} {status:<10} {added_at}")
            print("=" * 80)

        elif args.whitelist_action == "add":
            # 添加到白名单
            if not args.name:
                print("❌ 错误: 添加操作必须指定策略名称 (-n)")
                sys.exit(1)
            if manager.add_to_whitelist(args.name):
                print(f"✅ 成功添加到白名单: {args.name}")
            else:
                print(f"❌ 添加失败，策略可能已存在: {args.name}")
                sys.exit(1)

        elif args.whitelist_action == "remove":
            # 从白名单移除
            if not args.name:
                print("❌ 错误: 移除操作必须指定策略名称 (-n)")
                sys.exit(1)
            if manager.remove_from_whitelist(args.name):
                print(f"✅ 成功从白名单移除: {args.name}")
            else:
                print(f"❌ 移除失败，策略可能不在白名单中: {args.name}")
                sys.exit(1)

        elif args.whitelist_action == "enable":
            # 启用白名单中的策略
            if not args.name:
                print("❌ 错误: 启用操作必须指定策略名称 (-n)")
                sys.exit(1)
            if manager.set_whitelist_enabled(args.name, True):
                print(f"✅ 已启用: {args.name}")
            else:
                print(f"❌ 启用失败，策略可能不在白名单中: {args.name}")
                sys.exit(1)

        elif args.whitelist_action == "disable":
            # 禁用白名单中的策略
            if not args.name:
                print("❌ 错误: 禁用操作必须指定策略名称 (-n)")
                sys.exit(1)
            if manager.set_whitelist_enabled(args.name, False):
                print(f"✅ 已禁用: {args.name}")
            else:
                print(f"❌ 禁用失败，策略可能不在白名单中: {args.name}")
                sys.exit(1)

        elif args.whitelist_action == "auto-start":
            # 手动触发自动启动
            print("\n" + "=" * 80)
            print("🚀 手动触发白名单自动启动")
            print("=" * 80)

            results = manager.auto_start_whitelist_strategies()

            if not results:
                print("⚠️  今天已执行过自动启动或白名单为空")
            else:
                success_count = sum(1 for v in results.values() if v)
                fail_count = len(results) - success_count

                print(f"📊 结果: 成功 {success_count}, 失败 {fail_count}")
                for name, success in results.items():
                    status = "✅ 成功" if success else "❌ 失败"
                    print(f"  {status}: {name}")

            print("=" * 80)

        return

    # 原有逻辑
    if args.action == "status":
        status = manager.get_status()
        print_status_table(status)

    elif args.action in ["start", "stop"]:
        if args.all:
            manager.start_all() if args.action == "start" else manager.stop_all()
        elif args.name:
            manager.start_strategy(args.name) if args.action == "start" else manager.stop_strategy(args.name)
        else:
            print("❌ 错误: 请指定策略名称 (-n) 或使用 --all")
            sys.exit(1)

    elif args.action == "restart":
        if not args.name:
            print("❌ 错误: 重启操作必须指定策略名称 (-n)")
            sys.exit(1)
        manager.restart_strategy(args.name)

    elif args.action == "logs":
        if not args.name:
            print("❌ 错误: 查看日志必须指定策略名称 (-n)")
            sys.exit(1)

        log_file = Path("logs") / f"{args.name}.log"
        if not log_file.exists():
            print(f"⚠️  日志文件不存在: {log_file}")
            return

        print(f"\n📄 {args.name} 的日志 (最近{args.tail}行):")
        print("=" * 80)
        with open(log_file, 'r') as f:
            lines = f.readlines()[-args.tail:]
            for line in lines:
                print(line.rstrip())
        print("=" * 80)


if __name__ == "__main__":
    main()