technical-patterns-lab/scripts/pipeline_converging_triangle.py

"""
收敛三角形检测流水线

一键执行完整的检测、报告生成和图表绘制流程：
1. 运行批量检测 (run_converging_triangle.py)
2. 生成选股报告 (report_converging_triangles.py)
3. 绘制个股图表 (plot_converging_triangles.py)

用法:
    python scripts/pipeline_converging_triangle.py
    python scripts/pipeline_converging_triangle.py --date 20260120
    python scripts/pipeline_converging_triangle.py --show-details  # 生成详情模式图片
    python scripts/pipeline_converging_triangle.py --plot-boundary-source close  # 使用收盘价拟合边界线
"""

from __future__ import annotations

import argparse
import os
import sys
import time
from datetime import datetime

# 让脚本能找到 src/ 下的模块
sys.path.append(os.path.join(os.path.dirname(__file__), "..", "src"))

# 导入三个主要模块
from run_converging_triangle import main as run_detection
from report_converging_triangles import main as run_report
from plot_converging_triangles import main as run_plot
from generate_stock_viewer import main as run_viewer


def print_section_header(title: str, step: int) -> None:
    """打印流程步骤标题"""
    print("\n")
    print("=" * 80)
    print(f"步骤 {step}: {title}")
    print("=" * 80)
    print()


def print_step_result(success: bool, duration: float) -> None:
    """打印步骤执行结果"""
    status = "[完成]" if success else "[失败]"
    print()
    print("-" * 80)
    print(f"{status} | 耗时: {duration:.2f} 秒 ({duration/60:.2f} 分钟)")
    print("-" * 80)


def main() -> None:
    parser = argparse.ArgumentParser(description="收敛三角形检测完整流水线")
    parser.add_argument(
        "--date",
        type=int,
        default=None,
        help="指定日期（YYYYMMDD），用于报告和图表生成（默认为数据最新日）",
    )
    parser.add_argument(
        "--show-details",
        action="store_true",
        help="生成详情模式图片（显示所有枢轴点和拟合点）",
    )
    parser.add_argument(
        "--skip-detection",
        action="store_true",
        help="跳过批量检测步骤（如果已有检测结果）",
    )
    parser.add_argument(
        "--skip-report",
        action="store_true",
        help="跳过报告生成步骤",
    )
    parser.add_argument(
        "--skip-plot",
        action="store_true",
        help="跳过图表绘制步骤",
    )
    parser.add_argument(
        "--all-stocks",
        action="store_true",
        help="为所有108只股票生成图表（包括不满足收敛三角形条件的）",
    )
    parser.add_argument(
        "--skip-viewer",
        action="store_true",
        help="跳过生成HTML查看器",
    )
    parser.add_argument(
        "--clean",
        action="store_true",
        help="运行前清空outputs文件夹",
    )
    parser.add_argument(
        "--plot-boundary-source",
        choices=["hl", "close"],
        default="close",
        help="绘图时边界线拟合数据源: hl=高低价, close=收盘价（不影响检测）",
    )
    args = parser.parse_args()
    
    pipeline_start = time.time()
    
    print("=" * 80)
    print("收敛三角形检测流水线")
    print("=" * 80)
    print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    if args.date:
        print(f"指定日期: {args.date}")
    if args.show_details:
        print(f"图表模式: 详情模式（显示所有枢轴点）")
    else:
        print(f"图表模式: 简洁模式（仅显示价格和趋势线）")
    if args.all_stocks:
        print(f"图表范围: 所有108只股票（包括不满足条件的）")
    else:
        print(f"图表范围: 仅满足收敛三角形条件的股票")
    boundary_source_name = "收盘价" if args.plot_boundary_source == "close" else "高低价"
    print(f"边界线拟合数据源: {boundary_source_name}")
    print("=" * 80)
    
    # ========================================================================
    # 步骤 0: 清空输出目录（可选）
    # ========================================================================
    if args.clean:
        import shutil
        output_base = os.path.join(os.path.dirname(__file__), "..", "outputs", "converging_triangles")
        if os.path.exists(output_base):
            print("\n[清空输出目录]")
            print(f"  删除: {output_base}")
            shutil.rmtree(output_base)
            os.makedirs(output_base, exist_ok=True)
            print("  [OK] 输出目录已清空\n")
    
    results = []
    
    # ========================================================================
    # 步骤 1: 批量检测
    # ========================================================================
    if not args.skip_detection:
        print_section_header("批量检测 - 识别所有收敛三角形形态", 1)
        step_start = time.time()
        
        try:
            # 清除命令行参数，避免冲突
            sys.argv = [sys.argv[0]]
            run_detection()
            success = True
        except Exception as e:
            print(f"\n❌ 检测失败: {e}")
            success = False
        
        step_duration = time.time() - step_start
        print_step_result(success, step_duration)
        results.append(("批量检测", success, step_duration))
        
        if not success:
            print("\n[流水线中断：批量检测失败]")
            return
    else:
        print("\n[跳过批量检测步骤（使用已有结果）]")
        results.append(("批量检测", None, 0))
    
    # ========================================================================
    # 步骤 2: 生成报告
    # ========================================================================
    if not args.skip_report:
        print_section_header("生成报告 - 当日选股简报", 2)
        step_start = time.time()
        
        try:
            # 设置命令行参数
            if args.date:
                sys.argv = [sys.argv[0], "--report-date", str(args.date)]
            else:
                sys.argv = [sys.argv[0]]
            
            run_report()
            success = True
        except Exception as e:
            print(f"\n❌ 报告生成失败: {e}")
            success = False
        
        step_duration = time.time() - step_start
        print_step_result(success, step_duration)
        results.append(("生成报告", success, step_duration))
        
        if not success:
            print("\n[报告生成失败，但继续执行图表绘制]")
    else:
        print("\n[跳过报告生成步骤]")
        results.append(("生成报告", None, 0))
    
    # ========================================================================
    # 步骤 3: 绘制图表
    # ========================================================================
    if not args.skip_plot:
        print_section_header("绘制图表 - 个股收敛三角形可视化", 3)
        step_start = time.time()
        
        try:
            # 设置命令行参数
            cmd_args = [sys.argv[0]]
            if args.date:
                cmd_args.extend(["--date", str(args.date)])
            if args.show_details:
                cmd_args.append("--show-details")
            if args.all_stocks:
                cmd_args.append("--all-stocks")
            if args.plot_boundary_source:
                cmd_args.extend(["--plot-boundary-source", args.plot_boundary_source])
            
            sys.argv = cmd_args
            
            run_plot()
            success = True
        except Exception as e:
            print(f"\n❌ 图表绘制失败: {e}")
            success = False
        
        step_duration = time.time() - step_start
        print_step_result(success, step_duration)
        results.append(("绘制图表", success, step_duration))
    else:
        print("\n[跳过图表绘制步骤]")
        results.append(("绘制图表", None, 0))
    
    # ========================================================================
    # 步骤 4: 生成HTML查看器
    # ========================================================================
    if not args.skip_viewer:
        print_section_header("生成HTML查看器 - 可视化强度分筛选", 4)
        step_start = time.time()
        
        try:
            # 设置命令行参数
            cmd_args = [sys.argv[0]]
            if args.date:
                cmd_args.extend(["--date", str(args.date)])
            if args.all_stocks:
                cmd_args.append("--all-stocks")
            
            sys.argv = cmd_args
            
            run_viewer()
            success = True
        except Exception as e:
            print(f"\n❌ HTML查看器生成失败: {e}")
            success = False
        
        step_duration = time.time() - step_start
        print_step_result(success, step_duration)
        results.append(("生成查看器", success, step_duration))
    else:
        print("\n[跳过HTML查看器生成步骤]")
        results.append(("生成查看器", None, 0))
    
    # ========================================================================
    # 流水线总结
    # ========================================================================
    pipeline_duration = time.time() - pipeline_start
    
    print("\n")
    print("=" * 80)
    print("流水线执行总结")
    print("=" * 80)
    
    for step_name, success, duration in results:
        if success is None:
            status = "[跳过]"
            time_str = "-"
        elif success:
            status = "[成功]"
            time_str = f"{duration:.2f}s"
        else:
            status = "[失败]"
            time_str = f"{duration:.2f}s"
        
        print(f"{step_name:12} | {status:8} | {time_str:>10}")
    
    print("-" * 80)
    print(f"总耗时: {pipeline_duration:.2f} 秒 ({pipeline_duration/60:.2f} 分钟)")
    print(f"结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    # 检查是否所有步骤都成功
    all_success = all(s is None or s for _, s, _ in results)
    
    if all_success:
        print("\n[流水线执行完成]")
        print("\n输出文件:")
        print("  - outputs/converging_triangles/all_results.csv")
        print("  - outputs/converging_triangles/report.md")
        print("  - outputs/converging_triangles/charts/*.png")
        print("  - outputs/converging_triangles/stock_viewer.html  ← 用浏览器打开")
    else:
        print("\n[流水线部分失败，请检查上述错误信息]")
    
    print("=" * 80)


if __name__ == "__main__":
    main()