褚宏光
95d13b2cce
- Added `--show-details` parameter to `pipeline_converging_triangle.py` for generating detailed charts that display all pivot points and fitting lines. - Implemented an iterative outlier removal algorithm in `fit_pivot_line` to improve the accuracy of pivot point fitting by eliminating weak points. - Updated `USAGE.md` to include new command examples for the detailed mode. - Revised multiple documentation files to reflect recent changes and improvements in the pivot detection and visualization processes.
227 lines
7.3 KiB
Python
227 lines
7.3 KiB
Python
"""
|
||
收敛三角形检测流水线
|
||
|
||
一键执行完整的检测、报告生成和图表绘制流程：
|
||
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 # 生成详情模式图片
|
||
"""
|
||
|
||
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
|
||
|
||
|
||
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="跳过图表绘制步骤",
|
||
)
|
||
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"图表模式: 简洁模式（仅显示价格和趋势线）")
|
||
print("=" * 80)
|
||
|
||
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")
|
||
|
||
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))
|
||
|
||
# ========================================================================
|
||
# 流水线总结
|
||
# ========================================================================
|
||
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")
|
||
else:
|
||
print("\n[流水线部分失败，请检查上述错误信息]")
|
||
|
||
print("=" * 80)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|